Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
__author__ = 'facetoe'
class ZenpyException(Exception):
"""
A ``ZenpyException`` is raised for internal errors.
"""
class ZenpyCacheException(ZenpyException):
"""
A ``ZenpyCacheException`` is raised for errors relating the the :class:`ZenpyCache`.
"""
class RatelimitBudgetExceeded(ZenpyException):
"""
A ``RatelimitBudgetExceeded`` is raised when the ratelimit_budget has been spent.
"""
class APIException(Exception):
"""
An ``APIException`` is raised when the API rejects a query.
"""
def __init__(self, *args, **kwargs):
self.response = kwargs.pop('response', None)
super(APIException, self).__init__(*args)
class RecordNotFoundException(APIException):
def class_for_type(self, object_type):
if object_type not in self.class_mapping:
raise ZenpyException("Unknown object_type: " + str(object_type))
else:
return self.class_mapping[object_type]
def __call__(self, sort_order=None, sort_by=None, **kwargs):
if sort_order and sort_order not in ('asc', 'desc'):
raise ZenpyException("sort_order must be one of (asc, desc)")
if sort_by and sort_by not in ('alphabetical', 'created_at', 'updated_at', 'usage_1h', 'usage_24h', 'usage_7d'):
raise ZenpyException(
"sort_by is invalid - https://developer.zendesk.com/rest_api/docs/core/macros#available-parameters")
if 'id' in kwargs:
if len(kwargs) > 1:
raise ZenpyException("When specifying an id it must be the only parameter")
params = dict()
path = self.endpoint
for key, value in kwargs.items():
if isinstance(value, bool):
value = str(value).lower()
if key == 'id':
path += "/{}.json".format(value)
else:
params[key] = value
if sort_order:
params['sort_order'] = sort_order
if sort_by:
params['sort_by'] = sort_by
def class_for_type(self, object_type):
""" Given an object_type return the class associated with it. """
if object_type not in self.class_mapping:
raise ZenpyException("Unknown object_type: " + str(object_type))
else:
return self.class_mapping[object_type]
def put(self, endpoint, help_centre_object_id, translation):
if translation.locale is None:
raise ZenpyException("Locale can not be None when updating translation!")
url = self.api._build_url(endpoint(help_centre_object_id, translation.locale))
payload = self.build_payload(translation)
return self.api._put(url, payload=payload)
def _init_session(self, email, token, oath_token, password, session):
if not session:
session = requests.Session()
# Workaround for possible race condition - https://github.com/kennethreitz/requests/issues/3661
session.mount('https://', HTTPAdapter(**self.http_adapter_kwargs()))
if not hasattr(session, 'authorized') or not session.authorized:
# session is not an OAuth session that has been authorized, so authorize the session.
if not password and not token and not oath_token:
raise ZenpyException("password, token or oauth_token are required!")
elif password and token:
raise ZenpyException("password and token "
"are mutually exclusive!")
if password:
session.auth = (email, password)
elif token:
session.auth = ('%s/token' % email, token)
elif oath_token:
session.headers.update({'Authorization': 'Bearer %s' % oath_token})
else:
raise ZenpyException("Invalid arguments to _init_session()!")
headers = {'User-Agent': 'Zenpy/{}'.format(__version__)}
session.headers.update(headers)
return session
def __call__(self, start_time=None, **kwargs):
if start_time is None:
raise ZenpyException("Incremental Endpoint requires a start_time parameter!")
elif isinstance(start_time, datetime):
unix_time = to_unix_ts(start_time)
else:
unix_time = start_time
params = kwargs
params.update(dict(start_time=str(unix_time)))
if 'fields' in kwargs:
if is_iterable_but_not_string(kwargs['fields']):
f = ",".join(kwargs['fields'])
else:
f = kwargs['fields']
else:
f = "*"
def __call__(self, sort_order=None, sort_by=None, **kwargs):
if sort_order and sort_order not in ('asc', 'desc'):
raise ZenpyException("sort_order must be one of (asc, desc)")
if sort_by and sort_by not in ('alphabetical', 'created_at', 'updated_at', 'usage_1h', 'usage_24h', 'usage_7d'):
raise ZenpyException(
"sort_by is invalid - https://developer.zendesk.com/rest_api/docs/core/macros#available-parameters")
if 'id' in kwargs:
if len(kwargs) > 1:
raise ZenpyException("When specifying an id it must be the only parameter")
params = dict()
path = self.endpoint
for key, value in kwargs.items():
if isinstance(value, bool):
value = str(value).lower()
if key == 'id':
path += "/{}.json".format(value)
else:
if not hasattr(session, 'authorized') or not session.authorized:
# session is not an OAuth session that has been authorized, so authorize the session.
if not password and not token and not oath_token:
raise ZenpyException("password, token or oauth_token are required!")
elif password and token:
raise ZenpyException("password and token "
"are mutually exclusive!")
if password:
session.auth = (email, password)
elif token:
session.auth = ('%s/token' % email, token)
elif oath_token:
session.headers.update({'Authorization': 'Bearer %s' % oath_token})
else:
raise ZenpyException("Invalid arguments to _init_session()!")
headers = {'User-Agent': 'Zenpy/{}'.format(__version__)}
session.headers.update(headers)
return session
def _get_cache(self, cache_name):
if cache_name not in self.cache.mapping:
raise ZenpyException("No such cache - %s" % cache_name)
else:
return self.cache.mapping[cache_name]