Source code for apeye.slumber_url

#!/usr/bin/env python
#
#  slumber_url.py
"""
Subclass of :class:`~apeye.url.URL` with support for interacting with
REST APIs with `Slumber <https://slumber.readthedocs.io>`__ and
`Requests <https://requests.readthedocs.io>`__.

.. versionadded:: 0.2.0
"""  # noqa: D400
#
#  Copyright © 2020-2021 Dominic Davis-Foster <dominic@davis-foster.co.uk>
#
#  This program is free software; you can redistribute it and/or modify
#  it under the terms of the GNU Lesser General Public License as published by
#  the Free Software Foundation; either version 3 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
#  GNU Lesser General Public License for more details.
#
#  You should have received a copy of the GNU Lesser General Public License
#  along with this program; if not, write to the Free Software
#  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
#  MA 02110-1301, USA.
#
#  Based on the "pathlib" module from CPython.
#  Licensed under the Python Software Foundation License Version 2.
#  Copyright © 2001-2020 Python Software Foundation. All rights reserved.
#  Copyright © 2000 BeOpen.com. All rights reserved.
#  Copyright © 1995-2000 Corporation for National Research Initiatives. All rights reserved.
#  Copyright © 1991-1995 Stichting Mathematisch Centrum. All rights reserved.
#
#  Based on Slumber <https://slumber.readthedocs.io>
#  Copyright (c) 2011 Donald Stufft
#  Licensed under the 2-clause BSD License
#
#  Some docstrings from Requests <https://requests.readthedocs.io>
#  Copyright 2019 Kenneth Reitz
#  Licensed under the Apache License, Version 2.0
#

# stdlib
import copy
import sys
from typing import Callable, Dict, MutableMapping, Optional, Tuple, Union
from urllib.parse import unquote

# 3rd party
from requests import PreparedRequest, Session
from requests.auth import AuthBase
from requests.structures import CaseInsensitiveDict
from requests.utils import guess_json_utf

# this package
from apeye.requests_url import _Data
from apeye.slumber_url.exceptions import (
		HttpClientError,
		HttpNotFoundError,
		HttpServerError,
		SlumberBaseException,
		SlumberHttpBaseException
		)
from apeye.slumber_url.serializers import (
		JsonSerializer,
		Serializer,
		SerializerNotAvailable,
		SerializerRegistry,
		YamlSerializer
		)
from apeye.url import URL

__all__ = [
		"SlumberURL",
		"SerializerRegistry",
		"YamlSerializer",
		"Serializer",
		"JsonSerializer",
		"SlumberBaseException",
		"SlumberHttpBaseException",
		"HttpClientError",
		"HttpNotFoundError",
		"HttpServerError",
		"SerializerNotAvailable",
		]


# Ignore the LGTM warning as the "session" etc. attributes should **not** affect equality.
# Equality should only consider the URL, not its attributes.
[docs]class SlumberURL(URL): # lgtm [py/missing-equals] """ Subclass of :class:`~apeye.url.URL` with support for interacting with REST APIs with `Slumber <https://slumber.readthedocs.io>`__ and `Requests <https://requests.readthedocs.io>`__. :param url: The url to construct the :class:`~apeye.slumber_url.SlumberURL` object from. :param auth: :param format: :param append_slash: :param session: :param serializer: :param timeout: How long to wait for the server to send data before giving up. :param allow_redirects: Whether to allow redirects. :param proxies: Dictionary mapping protocol or protocol and hostname to the URL of the proxy. :param verify: Either a boolean, in which case it controls whether we verify the server's TLS certificate, or a string, in which case it must be a path to a CA bundle to use. :param cert: Either the path to the SSL client cert file (``.pem``), or a tuple of ``('cert', 'key')``. ``timeout``, ``allow_redirects``, ``proxies``, ``verify`` and ``cert`` are passed to Requests when making any HTTP requests, and are inherited by all children created from this URL. .. latex:vspace:: 10px .. versionchanged:: 0.3.0 The ``url`` parameter can now be a string or a :class:`~.URL`. .. versionchanged:: 1.1.0 When a :class:`~.RequestsURL` object is deleted or garbage collected, the underlying :class:`requests.Session` object it only closed if no objects hold references to the session. This prevents the session object of a global object from being inadvertently closed when one of its children is garbage collected. """ # noqa: D400 serializer: SerializerRegistry """ The serializer used to (de)serialize the data when interacting with the API. .. versionadded:: 0.6.0 """ session: Session """ The underlying requests session. .. versionadded:: 0.6.0 """ #: How long to wait for the server to send data before giving up. timeout: Union[None, float, Tuple[float, float], Tuple[float, None]] #: Whether to allow redirects. allow_redirects: Optional[bool] #: Dictionary mapping protocol or protocol and hostname to the URL of the proxy. proxies: Optional[MutableMapping[str, str]] verify: Union[None, bool, str] """ Either a boolean, in which case it controls whether we verify the server's TLS certificate, or a string, in which case it must be a path to a CA bundle to use. """ #: The path to ssl client cert file or a tuple of ``('cert', 'key')``. cert: Union[str, Tuple[str, str], None] def __init__( self, url: Union[str, URL] = '', auth: Union[None, Tuple[str, str], AuthBase, Callable[[PreparedRequest], PreparedRequest]] = None, format: str = "json", # noqa: A002 # pylint: disable=redefined-builtin append_slash: bool = True, session=None, serializer: Optional[SerializerRegistry] = None, *, timeout: Union[None, float, Tuple[float, float], Tuple[float, None]] = None, allow_redirects: Optional[bool] = True, proxies: Optional[MutableMapping[str, str]] = None, verify: Union[None, bool, str] = None, cert: Union[str, Tuple[str, str], None] = None, ): super().__init__(url) if serializer is None: serializer = SerializerRegistry(default=format) if session is None: session = Session() self.serializer = serializer self.session = session if auth is not None: self.session.auth = auth self._store = { "format": format if format is not None else "json", "append_slash": append_slash, "session": self.session, "serializer": self.serializer, } self.timeout = timeout self.allow_redirects = allow_redirects self.proxies = proxies self.verify = verify self.cert = cert
[docs] def url(self) -> str: """ Returns the URL as a string. """ url = str(self.base_url) if self._store["append_slash"] and not url.endswith('/'): url = url + '/' return url
def _request(self, method, data=None, files=None, params=None): serializer = self.serializer url = self.url() headers = {"accept": serializer.get_content_type()} if not files: headers["content-type"] = serializer.get_content_type() if data is not None: data = serializer.dumps(data) resp = self.session.request( method, url, data=data, params=params, files=files, headers=headers, timeout=self.timeout, allow_redirects=self.allow_redirects, # type: ignore[arg-type] proxies=self.proxies, verify=self.verify, cert=self.cert, ) if 400 <= resp.status_code <= 499: exception_class = HttpNotFoundError if resp.status_code == 404 else HttpClientError raise exception_class( f"Client Error {resp.status_code}: {unquote(resp.url)}", response=resp, content=resp.content, ) elif 500 <= resp.status_code <= 599: raise HttpServerError( f"Server Error {resp.status_code}: {unquote(resp.url)}", response=resp, content=resp.content, ) self._ = resp return resp def _try_to_serialize_response(self, resp): s = self.serializer if resp.status_code in [204, 205]: return if resp.headers.get("content-type", None) and resp.content: content_type = resp.headers.get("content-type").split(';')[0].strip() try: stype = s.get_serializer(content_type=content_type) except SerializerNotAvailable: return resp.content if isinstance(resp.content, bytes): try: encoding = guess_json_utf(resp.content) return stype.loads(resp.content.decode(encoding)) except Exception: return resp.content return stype.loads(resp.content) else: return resp.content def _process_response(self, resp): # TODO: something to expose headers and status if 200 <= resp.status_code <= 299: return self._try_to_serialize_response(resp) else: return # @@@ We should probably do some sort of error here? (Is this even possible?)
[docs] def get(self, **params) -> Dict: """ Perform a GET request using `Slumber <https://slumber.readthedocs.io>`__. https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/GET :param params: Parameters to send in the query string of the :class:`requests.Request`. """ resp = self._request("GET", params=params) return self._process_response(resp)
[docs] def post(self, data: _Data = None, files=None, **params) -> Dict: """ Perform a POST request using `Slumber <https://slumber.readthedocs.io>`__. https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/POST :param data: Dictionary, list of tuples, bytes, or file-like object to send in the body of the :class:`requests.Request`. :param files: Dictionary of ``'name': file-like-objects`` (or ``{'name': file-tuple}``) for multipart encoding upload. ``file-tuple`` can be a 2-tuple ``('filename', fileobj)``, 3-tuple ``('filename', fileobj, 'content_type')`` or a 4-tuple ``('filename', fileobj, 'content_type', custom_headers)``, where ``'content-type'`` is a string defining the content type of the given file and ``custom_headers`` a dict-like object containing additional headers to add for the file. :param params: Parameters to send in the query string of the :class:`requests.Request`. """ resp = self._request("POST", data=data, files=files, params=params) return self._process_response(resp)
[docs] def patch(self, data=None, files=None, **params) -> Dict: """ Perform a PATCH request using `Slumber <https://slumber.readthedocs.io>`__. https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/PATCH :param data: Dictionary, list of tuples, bytes, or file-like object to send in the body of the :class:`requests.Request`. :param files: Dictionary of ``'name': file-like-objects`` (or ``{'name': file-tuple}``) for multipart encoding upload. ``file-tuple`` can be a 2-tuple ``('filename', fileobj)``, 3-tuple ``('filename', fileobj, 'content_type')`` or a 4-tuple ``('filename', fileobj, 'content_type', custom_headers)``, where ``'content-type'`` is a string defining the content type of the given file and ``custom_headers`` a dict-like object containing additional headers to add for the file. :param params: Parameters to send in the query string of the :class:`requests.Request`. """ resp = self._request("PATCH", data=data, files=files, params=params) return self._process_response(resp)
[docs] def put(self, data=None, files=None, **params) -> Dict: """ Perform a PUT request using `Slumber <https://slumber.readthedocs.io>`__. https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/PUT :param data: Dictionary, list of tuples, bytes, or file-like object to send in the body of the :class:`requests.Request`. :param files: Dictionary of ``'name': file-like-objects`` (or ``{'name': file-tuple}``) for multipart encoding upload. ``file-tuple`` can be a 2-tuple ``('filename', fileobj)``, 3-tuple ``('filename', fileobj, 'content_type')`` or a 4-tuple ``('filename', fileobj, 'content_type', custom_headers)``, where ``'content-type'`` is a string defining the content type of the given file and ``custom_headers`` a dict-like object containing additional headers to add for the file. :param params: Parameters to send in the query string of the :class:`requests.Request`. """ resp = self._request("PUT", data=data, files=files, params=params) return self._process_response(resp)
[docs] def delete(self, **params) -> bool: """ Perform a DELETE request using `Slumber <https://slumber.readthedocs.io>`__. https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/DELETE :param params: Parameters to send in the query string of the :class:`requests.Request`. :returns: :py:obj:`True` if the DELETE request succeeded. :py:obj:`False` otherwise. """ resp = self._request("DELETE", params=params) # if 200 <= resp.status_code <= 299: # if resp.status_code == 204: # return True # else: # return True # else: # return False return 200 <= resp.status_code <= 299
[docs] def __del__(self): # pragma: no cover """ Attempt to close session when garbage collected to avoid leaving connections open. """ try: if sys.getrefcount(self.session) <= 2: self.session.close() except Exception: # nosec: B110 # pylint: disable=bare-except pass
[docs] def options(self, **kwargs) -> str: """ Send an OPTIONS request using `Requests <https://requests.readthedocs.io>`__. https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/OPTIONS :param kwargs: Optional arguments that :func:`requests.request` takes. """ return self.session.options(str(self.base_url), **kwargs).headers.get("Allow", '')
[docs] def head(self, **kwargs) -> CaseInsensitiveDict: """ Send a HEAD request using `Requests <https://requests.readthedocs.io>`__. https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/HEAD :param kwargs: Optional arguments that :func:`requests.request` takes. If `allow_redirects` is not provided, it will be set to :py:obj:`False` (as opposed to the default :func:`requests.request` behavior). """ return self.session.head(str(self.base_url), **kwargs).headers
def __truediv__(self, other): """ Construct a new :class:`~apeye.url.URL` object for the given child of this :class:`~apeye.url.URL`. """ new_obj = super().__truediv__(other) if new_obj is not NotImplemented: new_obj._store = copy.copy(self._store) new_obj.serializer = self.serializer new_obj.session = self.session new_obj.timeout = self.timeout new_obj.allow_redirects = self.allow_redirects new_obj.proxies = self.proxies new_obj.verify = self.verify new_obj.cert = self.cert return new_obj