Source code for uplink.returns

# Standard library imports
import sys
import warnings

# Local imports
from uplink import decorators
from uplink.converters import keys, interfaces

__all__ = ["json", "from_json", "schema"]


class ReturnType(object):
    def __init__(self, decorator, type_):
        self._decorator = decorator
        self._type_ = type_

    @property
    def type(self):
        return self._type_

    @staticmethod
    def with_decorator(old, decorator):
        old_type = None if old is None else old.type
        return ReturnType(decorator, decorator.return_type or old_type)

    def with_strategy(self, strategy):
        return CallableReturnType(self._decorator, self._type_, strategy)

    def is_applicable(self, decorator):
        return self._decorator is decorator


class CallableReturnType(ReturnType):
    def __init__(self, decorator, type_, strategy):
        super(CallableReturnType, self).__init__(decorator, type_)
        self._strategy = strategy

    def __call__(self, *args, **kwargs):
        return self._strategy(*args, **kwargs)


class _ReturnsBase(decorators.MethodAnnotation):
    @property
    def return_type(self):  # pragma: no cover
        raise NotImplementedError

    def _make_strategy(self, converter):  # pragma: no cover
        pass

    def _modify_request_definition(self, definition, kwargs):
        super(_ReturnsBase, self)._modify_request_definition(definition, kwargs)
        definition.return_type = ReturnType.with_decorator(
            definition.return_type, self
        )

    def _get_converter(self, request_builder, return_type):  # pragma: no cover
        return request_builder.get_converter(
            keys.CONVERT_FROM_RESPONSE_BODY, return_type.type
        )

    def modify_request(self, request_builder):
        return_type = request_builder.return_type
        if not return_type.is_applicable(self):
            return

        converter = self._get_converter(request_builder, return_type)
        if converter is None:
            return

        # Found a converter that can handle the return type.
        request_builder.return_type = return_type.with_strategy(
            self._make_strategy(converter)
        )


class JsonStrategy(object):
    # TODO: Consider moving this under json decorator
    # TODO: Support JSON Pointer (https://tools.ietf.org/html/rfc6901)

    def __init__(self, converter, key=()):
        self._converter = converter

        if not isinstance(key, (list, tuple)):
            key = (key,)
        self._key = key

    def __call__(self, response):
        content = response.json()
        for name in self._key:
            content = content[name]
        content = self._converter(content)
        return content


# noinspection PyPep8Naming
[docs]class json(_ReturnsBase): """ Specifies that the decorated consumer method should return a JSON object. .. code-block:: python # This method will return a JSON object (e.g., a dict or list) @returns.json @get("/users/{username}") def get_user(self, username): \"""Get a specific user.\""" Returning a Specific JSON Field: The :py:attr:`key` argument accepts a string or tuple that specifies the path of an internal field in the JSON document. For instance, consider an API that returns JSON responses that, at the root of the document, contains both the server-retrieved data and a list of relevant API errors: .. code-block:: json :emphasize-lines: 2 { "data": { "user": "prkumar", "id": "140232" }, "errors": [] } If returning the list of errors is unnecessary, we can use the :py:attr:`key` argument to strictly return the nested field :py:attr:`data.id`: .. code-block:: python @returns.json(key=("data", "id")) @get("/users/{username}") def get_user_id(self, username): \"""Get a specific user's ID.\""" We can also configure Uplink to convert the field before it's returned by also specifying the``type`` argument: .. code-block:: python @returns.json(key=("data", "id"), type=int) @get("/users/{username}") def get_user_id(self, username): \"""Get a specific user's ID.\""" .. versionadded:: v0.5.0 """ _can_be_static = True class _DummyConverter(interfaces.Converter): def convert(self, response): return response class _CastConverter(interfaces.Converter): def __init__(self, cast): self._cast = cast def convert(self, response): return self._cast(response) __dummy_converter = _DummyConverter() def __init__(self, type=None, key=(), model=None, member=()): if model: # pragma: no cover warnings.warn( "The `model` argument of @returns.json is deprecated and will " "be removed in v1.0.0. Use `type` instead.", DeprecationWarning, ) if member: # pragma: no cover warnings.warn( "The `member` argument of @returns.json is deprecated and will " "be removed in v1.0.0. Use `key` instead.", DeprecationWarning, ) self._type = type or model self._key = key or member @property def return_type(self): return self._type def _get_converter(self, request_builder, return_type): converter = super(json, self)._get_converter( request_builder, return_type ) if converter: return converter if callable(return_type.type): return self._CastConverter(return_type.type) # If the return_type cannot be converted, the strategy should directly # return the JSON body of the HTTP response, instead of trying to # deserialize it into a certain type. In this case, by # defaulting the return type to the dummy converter, which # implements this pass-through behavior, we ensure that # _make_strategy is called. return self.__dummy_converter def _make_strategy(self, converter): return JsonStrategy(converter, self._key)
from_json = json """ Specifies that the decorated consumer method should produce instances of a :py:obj:`type` class using a registered deserialization strategy (see :py:meth:`uplink.loads.from_json`) This decorator accepts the same arguments as :py:class:`uplink.returns.json`. Often, a JSON response body represents a schema in your application. If an existing Python object encapsulates this schema, use the :py:attr:`type` argument to specify it as the return type: .. code-block:: python @returns.from_json(type=User) @get("/users/{username}") def get_user(self, username): \"""Get a specific user.\""" For Python 3 users, you can alternatively provide a return value annotation. Hence, the previous code is equivalent to the following in Python 3: .. code-block:: python @returns.from_json @get("/users/{username}") def get_user(self, username) -> User: \"""Get a specific user.\""" Both usages typically require also registering a converter that knows how to deserialize the JSON into the specified :py:attr:`type` (see :py:meth:`uplink.loads.from_json`). This step is unnecessary if the :py:attr:`type` is defined using a library for which Uplink has built-in support, such as :py:mod:`marshmallow`. .. versionadded:: v0.6.0 """ # noinspection PyPep8Naming
[docs]class schema(_ReturnsBase): """ Specifies that the function returns a specific type of response. In Python 3, to provide a consumer method's return type, you can set it as the method's return annotation: .. code-block:: python @get("/users/{username}") def get_user(self, username) -> UserSchema: \"""Get a specific user.\""" For Python 2.7 compatibility, you can use this decorator instead: .. code-block:: python @returns.schema(UserSchema) @get("/users/{username}") def get_user(self, username): \"""Get a specific user.\""" To have Uplink convert response bodies into the desired type, you will need to define an appropriate converter (e.g., using :py:class:`uplink.loads`). .. versionadded:: v0.5.1 """ def __init__(self, type): self._schema = type @property def return_type(self): return self._schema def _make_strategy(self, converter): return converter
class _ModuleProxy(object): __module = sys.modules[__name__] schema = model = schema json = json from_json = from_json __all__ = __module.__all__ def __getattr__(self, item): return getattr(self.__module, item) def __call__(self, *args, **kwargs): return schema(*args, **kwargs) sys.modules[__name__] = _ModuleProxy()