Source code for contentful_management.utils

import re
import sys
import time
import json
from random import uniform
from .errors import RateLimitExceededError

import logging
try:  # Python 2.7+
    from logging import NullHandler
except ImportError:
    class NullHandler(logging.Handler):
        def emit(self, record):
            pass

"""
contentful_management.utils
~~~~~~~~~~~~~~~~~~~~~~~~~~~

This module implements utilities.

:copyright: (c) 2018 by Contentful GmbH.
:license: MIT, see LICENSE for more details.
"""


logging.getLogger(__name__).addHandler(NullHandler())
log = logging.getLogger(__name__)


[docs]def unicode_class(): """ Returns the class that allows for unicode encoded strings depends on the Python version. """ if sys.version_info[0] >= 3: return str return unicode # noqa: F821
[docs]def string_class(): """Returns the parent class for strings depends on the Python version.""" if sys.version_info[0] >= 3: return str return basestring
[docs]def json_error_class(): """Returns the class for JSON decode errors depends on the Python version.""" if sys.version_info[0] >= 3 and sys.version_info[1] >= 5: return json.JSONDecodeError return ValueError
[docs]def snake_case(a_string): """ Returns a snake-cased version of a string. :param a_string: any :class:`str` object. Usage: >>> snake_case('FooBar') "foo_bar" """ partial = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', a_string) return re.sub('([a-z0-9])([A-Z])', r'\1_\2', partial).lower()
[docs]def camel_case(snake_str): """ Returns a camel-cased version of a string. :param a_string: any :class:`str` object. Usage: >>> camel_case('foo_bar') "fooBar" """ components = snake_str.split('_') # We capitalize the first letter of each component except the first one # with the 'title' method and join them together. return components[0] + "".join(x.title() for x in components[1:])
[docs]def base_path_for(resource_name): """ Returns the path for a specified resource name. """ return { 'Tag': 'tags', 'Role': 'roles', 'Space': 'spaces', 'Asset': 'assets', 'Entry': 'entries', 'Locale': 'locales', 'Upload': 'uploads', 'ApiKey': 'api_keys', 'UIExtension': 'extensions', 'ContentType': 'content_types', 'Webhook': 'webhook_definitions', 'PreviewApiKey': 'preview_api_keys', 'SpaceMembership': 'space_memberships' }[resource_name]
[docs]def normalize_select(query): """ If the query contains the :select operator, we enforce :sys properties. The SDK requires sys.type to function properly, but as other of our SDKs require more parts of the :sys properties, we decided that every SDK should include the complete :sys block to provide consistency accross our SDKs. """ if 'select' not in query: return if isinstance( query['select'], str_type()): query['select'] = [s.strip() for s in query['select'].split(',')] query['select'] = [s for s in query['select'] if not s.startswith('sys.')] if 'sys' not in query['select']: query['select'].append('sys')
[docs]def str_type(): """ Returns the correct string type for the current Python version. """ if sys.version_info[0] >= 3: return str global basestring return basestring
[docs]def sanitize_date(date): """In order to have a more accurate comparison due to minimal delays upon publishing entries. We strip milliseconds from the dates we compare. :param date: datetime.datetime :return: datetime.datetime without milliseconds. """ return date.replace(microsecond=0)
[docs]class ConfigurationException(Exception): """ Configuration error class. """ pass
[docs]class retry_request(object): """ Decorator to retry function calls in case they raise rate limit exceptions. """ RATE_LIMIT_RESET_HEADER_KEY = 'x-contentful-ratelimit-reset' def __init__(self, client): self.client = client def __call__(self, http_call): def wrapper(url, query=None, **kwargs): exception = None for i in range(self.client.max_rate_limit_retries + 1): try: return http_call(url, query, **kwargs) except RateLimitExceededError as error: exception = error reset_time = error.reset_time() if reset_time > self.client.max_rate_limit_wait: raise error retry_message = 'Contentful API Rate Limit Hit! ' retry_message += "Retrying - Retries left: {0} ".format( self.client.max_rate_limit_retries - i ) retry_message += "- Time until reset (seconds): {0}".format( reset_time ) log.debug(retry_message) time.sleep(reset_time * uniform(1.0, 1.2)) raise exception return wrapper