PK M/pygetty/__init__.pyfrom __future__ import unicode_literals __doc__ = 'Wrapper for Getty v3 API' __author__ = 'Josh Klar ' __version__ = '1.0.0' PK M# pygetty/auth.pyfrom __future__ import absolute_import, unicode_literals from builtins import str import pendulum import requests from .util import gen_url TOKEN_VALIDITY_PADDING = 5 # trim these many seconds off token validity GRANT_TYPE_CLIENT_CREDENTIALS = 'client_credentials' class AuthToken(object): def __init__(self, access_token, expires_in=1800, token_type='Bearer'): if not access_token: raise ValueError('access_token must be provided and truthy') self.access_token = access_token self.expiry = pendulum.now().add(seconds=int(expires_in)) self.token_type = token_type def __repr__(self): return '<{}AuthToken: {}>'.format( '' if self.valid else 'Expired ', str(self), ) def __str__(self): return str(self.access_token) def __unicode__(self): return str(self.access_token) @staticmethod def from_dict(obj): return AuthToken(**obj) @property def valid(self): """ Ensures a token is valid now, and for at least the next few seconds. """ return self.expiry > pendulum.now().add(seconds=TOKEN_VALIDITY_PADDING) class AuthTokenManager(object): def __init__(self, api_key, client_secret, auth_token=None): self.api_key = api_key self.client_secret = client_secret self.auth_token = auth_token def _fetch_token(self): res = requests.post(gen_url('oauth2', 'token'), data={ 'grant_type': GRANT_TYPE_CLIENT_CREDENTIALS, 'client_id': self.api_key, 'client_secret': self.client_secret, }) res.raise_for_status() return AuthToken.from_dict(res.json()) @property def token(self): """ A wrapper around AuthTokenManager.auth_token which will always return a currently-valid token (or raise a requests Exception) """ if not self.auth_token or not self.auth_token.valid: self.auth_token = self._fetch_token() return self.auth_token def request_headers(self): return { 'Api-Key': self.api_key, 'Authorization': 'Bearer {}'.format(self.token), } def flex_auth(api_key=None, client_secret=None, auth_token_manager=None): """ Takes either an AuthTokenManager (which is passed through), or an API Key and Client Secret (which is turned into an AuthTokenManager). Exists so endpoint wrappers can take either an ATM or raw creds at the call validation level, but only need to handle ATMs in the "real" functionality. This entire flow basically exists to make the REPL and one-off calls less boilerplatey. """ if auth_token_manager: return auth_token_manager if not (api_key and client_secret): raise ValueError('Either auth_token_manager or api_key+client_secret required') return AuthTokenManager(api_key, client_secret) PK Mcÿpygetty/consts.pyfrom __future__ import unicode_literals base_url = 'https://api.gettyimages.com' base_url_v3 = 'https://api.gettyimages.com/v3' PK M`[pygetty/downloads.pyfrom __future__ import absolute_import, unicode_literals from builtins import str import requests from .auth import flex_auth from .util import gen_v3_url def image_download_url( id, file_type='jpg', auth_token_manager=None, api_key=None, client_secret=None, ): """ Request the URL from which to download a full-resolution Getty asset. """ auth_token_manager = flex_auth( api_key=api_key, client_secret=client_secret, auth_token_manager=auth_token_manager, ) res = requests.post( gen_v3_url('downloads', 'images', str(id)), headers=auth_token_manager.request_headers(), params={ 'auto_download': False, 'file_type': file_type, }, ) res.raise_for_status() return res.json().get('uri') def video_download_url( id, size='hd1', auth_token_manager=None, api_key=None, client_secret=None, ): """ Request the URL from which to download a full-resolution Getty asset. """ auth_token_manager = flex_auth( api_key=api_key, client_secret=client_secret, auth_token_manager=auth_token_manager, ) res = requests.post( gen_v3_url('downloads', 'videos', str(id)), headers=auth_token_manager.request_headers(), params={ 'auto_download': False, 'size': size, }, ) res.raise_for_status() return res.json().get('uri') PK Mt;DVpygetty/formatters.pyfrom __future__ import absolute_import, print_function, unicode_literals import re import pendulum MASTERY_DIMENSIONS_REGEX = re.compile(r'(?P[0-9]+)x(?P[0-9]+)') def format_video(video): if 'date_created' in video: video['date_created'] = pendulum.parse(video['date_created']) if 'clip_length' in video: cl = [int(x) for x in video['clip_length'].split(':')] video['clip_length'] = pendulum.duration( days=cl[0], hours=cl[1], minutes=cl[2], seconds=cl[3], ) if 'keywords' in video: video['raw_keywords'] = video['keywords'] video['keywords'] = [kw['text'] for kw in video['keywords']] if 'mastered_to' in video: video['parsed_dimensions'] = { k: int(v) for k, v in re.search( pattern=MASTERY_DIMENSIONS_REGEX, string=video['mastered_to'], ).groupdict().viewitems() } return video PK M5 MAX_PAGE_SIZE: warnings.warn(dedent(""" search: Requested page_size {page_size} is greater than max {max_page_size}, using {max_page_size} """).format( page_size=page_size, max_page_size=MAX_PAGE_SIZE, ), APIPageSizeLimitExceeded) auth_token_manager = flex_auth( auth_token_manager=auth_token_manager, api_key=api_key, client_secret=client_secret, ) returned = 0 page_num = start_page page_size = min((page_size, MAX_PAGE_SIZE)) params = deepcopy(query_params) new_fields = fields.copy() if detailed: new_fields.add('detail_set') params['fields'] = ','.join(new_fields) while returned < max_results: try: page = _fetch_page( page=page_num, page_size=page_size, query_params=params, asset_type=asset_type, search_type=search_type, auth_token_manager=auth_token_manager, ) except requests.exceptions.HTTPError as e: if e.response.status_code == 400 and 'page must be equal to' in e.response.text: logger.warning('Got page must be equal to error') return raise for asset in page[asset_type]: yield asset_formatters[asset_type](asset) returned += 1 if returned >= max_results: return if len(page[asset_type]) < page_size: return page_num += 1 def creative_videos(*args, **kwargs): kwargs['search_type'] = 'creative' kwargs['asset_type'] = 'videos' return search(*args, **kwargs) def editorial_videos(*args, **kwargs): kwargs['search_type'] = 'editorial' kwargs['asset_type'] = 'videos' return search(*args, **kwargs) def creative_images(*args, **kwargs): kwargs['search_type'] = 'creative' kwargs['asset_type'] = 'images' return search(*args, **kwargs) def editorial_images(*args, **kwargs): kwargs['search_type'] = 'editorial' kwargs['asset_type'] = 'images' return search(*args, **kwargs) PK M811pygetty/usage.pyfrom __future__ import absolute_import, unicode_literals from builtins import str import pendulum import requests from .auth import flex_auth from .util import gen_v3_url class ReportingFailed(Exception): def __init__(self, message, invalid_assets): super(ReportingFailed, self).__init__(message) self.invalid_assets = invalid_assets def report_usage_now( id, api_key=None, client_secret=None, auth_token_manager=None, ): """ Reports asset usage at the current moment for a single asset. Returns 1 on success, or raises ReportingFailed. """ auth_token_manager = flex_auth( api_key=api_key, client_secret=client_secret, auth_token_manager=auth_token_manager, ) res = requests.put( gen_v3_url('usage-batches', str(id)), headers=auth_token_manager.request_headers(), json={ 'asset_usages': [{ 'asset_id': id, 'quantity': 1, 'usage_date': pendulum.now().isoformat(), }], }, ) res.raise_for_status() if res.json().get('invalid_assets'): raise ReportingFailed( 'Some assets failed to report', res.json()['invalid_assets'], ) return res.json()['total_asset_usages_processed'] PK M6Jpygetty/util.pyfrom __future__ import absolute_import, unicode_literals from .consts import base_url, base_url_v3 def gen_url(*args): """ Generate an "old"-style URL. As far as I know this is only needed for auth. """ join_args = [base_url] join_args.extend(args) return '/'.join(join_args) def gen_v3_url(*args): """ Generate a v3 REST API URL. """ join_args = [base_url_v3] join_args.extend(args) return '/'.join(join_args) PK M9̹pygetty-1.0.0.dist-info/LICENSEISC License Copyright 2018 Sniply Projects, Inc. (d/b/a Lumen5) Permission to use, copy, modify, and/or distribute this software for any purpose with or without fee is hereby granted, provided that the above copyright notice and this permission notice appear in all copies. THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. PK!Hp!Qapygetty-1.0.0.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,zd&Y)r$[)T&UD"PK!H^N pygetty-1.0.0.dist-info/METADATA]=O0waGA XB"*Ag\86s,vT:t|YU䵝$\+Q+#,N*\4E)琠KPL#c`v^fYyޚ ^=N>VMբyО%Pܞ]@x:Y6PK!H C6upygetty-1.0.0.dist-info/RECORDun@}'LĀ ޔ0cf΢ ?ҽsl`k;-4;>wD`KW-Ϊrw'V-XGEC_{*n|RkkrR/#ތP]X81jpWR8k5c{ 9u.߄s(~E PƦBQty8^uPRUkJg Iy+cQߧc 2 m2qk6E2MP$pu[dt})defe~svM/(*&.!ԏ4(OWҝ8s; F C;I Y5Z9#^ |afe24^:G'^ɯK 7ЛQ^`Dt:}4*^ہS3[۩<SǞ5E1gs6hYB[.%,_0cުmakJuB5Melbǎ4O&6=C|&)@@9q]JLa_ݓ?dx PK M/pygetty/__init__.pyPK M# pygetty/auth.pyPK Mcÿ pygetty/consts.pyPK M`[0 pygetty/downloads.pyPK Mt;DVpygetty/formatters.pyPK M5