diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/python_http_client/client.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/python_http_client/client.py | 296 |
1 files changed, 296 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/python_http_client/client.py b/.venv/lib/python3.12/site-packages/python_http_client/client.py new file mode 100644 index 00000000..3a562351 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/python_http_client/client.py @@ -0,0 +1,296 @@ +"""HTTP Client library""" +import json +import logging +from .exceptions import handle_error + +try: + # Python 3 + import urllib.request as urllib + from urllib.parse import urlencode + from urllib.error import HTTPError +except ImportError: + # Python 2 + import urllib2 as urllib + from urllib2 import HTTPError + from urllib import urlencode + +_logger = logging.getLogger(__name__) + + +class Response(object): + """Holds the response from an API call.""" + + def __init__(self, response): + """ + :param response: The return value from a open call + on a urllib.build_opener() + :type response: urllib response object + """ + self._status_code = response.getcode() + self._body = response.read() + self._headers = response.info() + + @property + def status_code(self): + """ + :return: integer, status code of API call + """ + return self._status_code + + @property + def body(self): + """ + :return: response from the API + """ + return self._body + + @property + def headers(self): + """ + :return: dict of response headers + """ + return self._headers + + @property + def to_dict(self): + """ + :return: dict of response from the API + """ + if self.body: + return json.loads(self.body.decode('utf-8')) + else: + return None + + +class Client(object): + """Quickly and easily access any REST or REST-like API.""" + + # These are the supported HTTP verbs + methods = {'delete', 'get', 'patch', 'post', 'put'} + + def __init__(self, + host, + request_headers=None, + version=None, + url_path=None, + append_slash=False, + timeout=None): + """ + :param host: Base URL for the api. (e.g. https://api.sendgrid.com) + :type host: string + :param request_headers: A dictionary of the headers you want + applied on all calls + :type request_headers: dictionary + :param version: The version number of the API. + Subclass _build_versioned_url for custom behavior. + Or just pass the version as part of the URL + (e.g. client._("/v3")) + :type version: integer + :param url_path: A list of the url path segments + :type url_path: list of strings + """ + self.host = host + self.request_headers = request_headers or {} + self._version = version + # _url_path keeps track of the dynamically built url + self._url_path = url_path or [] + # APPEND SLASH set + self.append_slash = append_slash + self.timeout = timeout + + def _build_versioned_url(self, url): + """Subclass this function for your own needs. + Or just pass the version as part of the URL + (e.g. client._('/v3')) + :param url: URI portion of the full URL being requested + :type url: string + :return: string + """ + return '{}/v{}{}'.format(self.host, str(self._version), url) + + def _build_url(self, query_params): + """Build the final URL to be passed to urllib + + :param query_params: A dictionary of all the query parameters + :type query_params: dictionary + :return: string + """ + url = '' + count = 0 + while count < len(self._url_path): + url += '/{}'.format(self._url_path[count]) + count += 1 + + # add slash + if self.append_slash: + url += '/' + + if query_params: + url_values = urlencode(sorted(query_params.items()), True) + url = '{}?{}'.format(url, url_values) + + if self._version: + url = self._build_versioned_url(url) + else: + url = '{}{}'.format(self.host, url) + return url + + def _update_headers(self, request_headers): + """Update the headers for the request + + :param request_headers: headers to set for the API call + :type request_headers: dictionary + :return: dictionary + """ + self.request_headers.update(request_headers) + + def _build_client(self, name=None): + """Make a new Client object + + :param name: Name of the url segment + :type name: string + :return: A Client object + """ + url_path = self._url_path + [name] if name else self._url_path + return Client(host=self.host, + version=self._version, + request_headers=self.request_headers, + url_path=url_path, + append_slash=self.append_slash, + timeout=self.timeout) + + def _make_request(self, opener, request, timeout=None): + """Make the API call and return the response. This is separated into + it's own function, so we can mock it easily for testing. + + :param opener: + :type opener: + :param request: url payload to request + :type request: urllib.Request object + :param timeout: timeout value or None + :type timeout: float + :return: urllib response + """ + timeout = timeout or self.timeout + try: + return opener.open(request, timeout=timeout) + except HTTPError as err: + exc = handle_error(err) + exc.__cause__ = None + _logger.debug('{method} Response: {status} {body}'.format( + method=request.get_method(), + status=exc.status_code, + body=exc.body)) + raise exc + + def _(self, name): + """Add variable values to the url. + (e.g. /your/api/{variable_value}/call) + Another example: if you have a Python reserved word, such as global, + in your url, you must use this method. + + :param name: Name of the url segment + :type name: string + :return: Client object + """ + return self._build_client(name) + + def __getattr__(self, name): + """Dynamically add method calls to the url, then call a method. + (e.g. client.name.name.method()) + You can also add a version number by using .version(<int>) + + :param name: Name of the url segment or method call + :type name: string or integer if name == version + :return: mixed + """ + if name == 'version': + def get_version(*args, **kwargs): + """ + :param args: dict of settings + :param kwargs: unused + :return: string, version + """ + self._version = args[0] + return self._build_client() + return get_version + + # We have reached the end of the method chain, make the API call + if name in self.methods: + method = name.upper() + + def http_request( + request_body=None, + query_params=None, + request_headers=None, + timeout=None, + **_): + """Make the API call + :param timeout: HTTP request timeout. Will be propagated to + urllib client + :type timeout: float + :param request_headers: HTTP headers. Will be merged into + current client object state + :type request_headers: dict + :param query_params: HTTP query parameters + :type query_params: dict + :param request_body: HTTP request body + :type request_body: string or json-serializable object + :param kwargs: + :return: Response object + """ + if request_headers: + self._update_headers(request_headers) + + if request_body is None: + data = None + else: + # Don't serialize to a JSON formatted str + # if we don't have a JSON Content-Type + if 'Content-Type' in self.request_headers and \ + self.request_headers['Content-Type'] != \ + 'application/json': + data = request_body.encode('utf-8') + else: + self.request_headers.setdefault( + 'Content-Type', 'application/json') + data = json.dumps(request_body).encode('utf-8') + + opener = urllib.build_opener() + request = urllib.Request( + self._build_url(query_params), + headers=self.request_headers, + data=data, + ) + request.get_method = lambda: method + + _logger.debug('{method} Request: {url}'.format( + method=method, + url=request.get_full_url())) + if request.data: + _logger.debug('PAYLOAD: {data}'.format( + data=request.data)) + _logger.debug('HEADERS: {headers}'.format( + headers=request.headers)) + + response = Response( + self._make_request(opener, request, timeout=timeout) + ) + + _logger.debug('{method} Response: {status} {body}'.format( + method=method, + status=response.status_code, + body=response.body)) + + return response + + return http_request + else: + # Add a segment to the URL + return self._(name) + + def __getstate__(self): + return self.__dict__ + + def __setstate__(self, state): + self.__dict__ = state |