Source code for twilio.base.version

import json
from math import ceil

from twilio.base import values
from twilio.base.exceptions import TwilioRestException


[docs]class Version(object): """ Represents an API version. """ def __init__(self, domain): """ :param Domain domain: :return: """ self.domain = domain self.version = None
[docs] def absolute_url(self, uri): """ Turns a relative uri into an absolute url. """ return self.domain.absolute_url(self.relative_uri(uri))
[docs] def relative_uri(self, uri): """ Turns a relative uri into a versioned relative uri. """ return '{}/{}'.format(self.version.strip('/'), uri.strip('/'))
[docs] def request(self, method, uri, params=None, data=None, headers=None, auth=None, timeout=None, allow_redirects=False): """ Make an HTTP request. """ url = self.relative_uri(uri) return self.domain.request( method, url, params=params, data=data, headers=headers, auth=auth, timeout=timeout, allow_redirects=allow_redirects )
[docs] @classmethod def exception(cls, method, uri, response, message): """ Wraps an exceptional response in a `TwilioRestException`. """ # noinspection PyBroadException try: error_payload = json.loads(response.text) if 'message' in error_payload: message = '{}: {}'.format(message, error_payload['message']) details = error_payload.get('details') code = error_payload.get('code', response.status_code) return TwilioRestException(response.status_code, uri, message, code, method, details) except Exception: return TwilioRestException(response.status_code, uri, message, response.status_code, method)
[docs] def fetch(self, method, uri, params=None, data=None, headers=None, auth=None, timeout=None, allow_redirects=False): """ Fetch a resource instance. """ response = self.request( method, uri, params=params, data=data, headers=headers, auth=auth, timeout=timeout, allow_redirects=allow_redirects, ) # Note that 3XX response codes are allowed for fetches. if response.status_code < 200 or response.status_code >= 400: raise self.exception(method, uri, response, 'Unable to fetch record') return json.loads(response.text)
[docs] def update(self, method, uri, params=None, data=None, headers=None, auth=None, timeout=None, allow_redirects=False): """ Update a resource instance. """ response = self.request( method, uri, params=params, data=data, headers=headers, auth=auth, timeout=timeout, allow_redirects=allow_redirects, ) if response.status_code < 200 or response.status_code >= 300: raise self.exception(method, uri, response, 'Unable to update record') return json.loads(response.text)
[docs] def delete(self, method, uri, params=None, data=None, headers=None, auth=None, timeout=None, allow_redirects=False): """ Delete a resource. """ response = self.request( method, uri, params=params, data=data, headers=headers, auth=auth, timeout=timeout, allow_redirects=allow_redirects, ) if response.status_code < 200 or response.status_code >= 300: raise self.exception(method, uri, response, 'Unable to delete record') return response.status_code == 204
[docs] def read_limits(self, limit=None, page_size=None): """ Takes a limit on the max number of records to read and a max page_size and calculates the max number of pages to read. :param int limit: Max number of records to read. :param int page_size: Max page size. :return dict: A dictionary of paging limits. """ if limit is not None and page_size is None: page_size = limit return { 'limit': limit or values.unset, 'page_size': page_size or values.unset, }
[docs] def page(self, method, uri, params=None, data=None, headers=None, auth=None, timeout=None, allow_redirects=False): """ Makes an HTTP request. """ return self.request( method, uri, params=params, data=data, headers=headers, auth=auth, timeout=timeout, allow_redirects=allow_redirects, )
[docs] def stream(self, page, limit=None, page_limit=None): """ Generates records one a time from a page, stopping at prescribed limits. :param Page page: The page to stream. :param int limit: The max number of records to read. :param int page_limit: The max number of pages to read. """ current_record = 1 current_page = 1 while page is not None: for record in page: yield record current_record += 1 if limit and limit is not values.unset and limit < current_record: return current_page += 1 if page_limit and page_limit is not values.unset and page_limit < current_page: return page = page.next_page()
[docs] def create(self, method, uri, params=None, data=None, headers=None, auth=None, timeout=None, allow_redirects=False): """ Create a resource instance. """ response = self.request( method, uri, params=params, data=data, headers=headers, auth=auth, timeout=timeout, allow_redirects=allow_redirects, ) if response.status_code < 200 or response.status_code >= 300: raise self.exception(method, uri, response, 'Unable to create record') return json.loads(response.text)