PKY9M_kpygetty/__init__.pyfrom __future__ import unicode_literals __doc__ = 'Wrapper for Getty v3 API' __author__ = 'Josh Klar ' __version__ = '1.1.3' PKY9M# 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) PKY9Mcÿpygetty/consts.pyfrom __future__ import unicode_literals base_url = 'https://api.gettyimages.com' base_url_v3 = 'https://api.gettyimages.com/v3' PKY9M`[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') PKY9MvOhhpygetty/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_image(image): if image.get('date_created') is not None: image['date_created'] = pendulum.parse(image['date_created']) if image.get('keywords') is not None: image['raw_keywords'] = image['keywords'] image['keywords'] = [kw['text'] for kw in image['keywords']] return image def format_video(video): if video.get('date_created') is not None: video['date_created'] = pendulum.parse(video['date_created']) if video.get('clip_length') is not None: 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 video.get('keywords') is not None: video['raw_keywords'] = video['keywords'] video['keywords'] = [kw['text'] for kw in video['keywords']] if video.get('mastered_to') is not None: video['parsed_dimensions'] = { k: int(v) for k, v in re.search( pattern=MASTERY_DIMENSIONS_REGEX, string=video['mastered_to'], ).groupdict().viewitems() } return video PKY9Mpygetty/search.pyfrom __future__ import absolute_import, unicode_literals import logging import warnings from copy import deepcopy from textwrap import dedent import requests from .auth import flex_auth from .formatters import format_image, format_video from .util import gen_v3_url DEFAULT_PAGE_SIZE = 30 MAX_PAGE_SIZE = 100 DEFAULT_MAX_RESULTS = float('inf') logger = logging.getLogger(__name__) asset_formatters = { 'videos': format_video, 'images': format_image, } class APIPageSizeLimitExceeded(Warning): pass def _fetch_page( page, page_size, query_params, asset_type, search_type=None, auth_token_manager=None, ): params = { 'page': page, 'page_size': page_size, } params.update(query_params) if search_type: url = gen_v3_url('search', asset_type, search_type) else: url = gen_v3_url('search', asset_type) res = requests.get( url, headers=auth_token_manager.request_headers(), params=params, ) res.raise_for_status() return res.json() def search( max_results=DEFAULT_MAX_RESULTS, start_page=1, page_size=DEFAULT_PAGE_SIZE, detailed=False, fields=set(), query_params={}, asset_type=None, search_type=None, auth_token_manager=None, api_key=None, client_secret=None, ): if page_size > 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 all_videos(*args, **kwargs): kwargs['asset_type'] = 'videos' return search(*args, **kwargs) 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 all_images(*args, **kwargs): kwargs['asset_type'] = 'images' 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) PKY9M/x pygetty/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() PKY9M6Jpygetty/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) PKY9M9̹pygetty-1.1.3.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.1.3.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,zd&Y)r$[)T&UD"PK!HQN pygetty-1.1.3.dist-info/METADATA]=O0waHBDT$ )"Φ$FqlJY8t~=: jk;INsQ+#,NQJ4E)琠K+(^Kd &N10;/FRl6S&t7JCν#Ox؅ԃ|PhQ8ft8 壋lW8vz>prc~7mKlpΑk0%g\ $  \n-E7ycbn1r;՚JZGX\w? Qռ.d\b6Reڶ 暜.b?z|4s