Handling multiple return type

[EvaSDK]

Hello, I have an API that returns different JSON structure according to whether an operation was successful or failed. I am wrapping the API in a command line tool and I would like to handle deserialization with uplink and not raise exceptions but simply print the error message returned.

Here is a quick excerpt of what it looks like currently:

class Success(BaseModel):
    code: conint(200, 299)
    message: str


class Failure(BaseModel):
    errors: Dict


def verbose_response(resp):
    ...
    if "errors" in resp.content:
        print("Error handling")

    return resp


@response_handler(verbose_response)
class MyApi(Consumer):

    @post("/resource")
    def create_resource(self, **Body(type=MyPayload) -> Union[Success, Failure]:
        ...

Unfortunately, this raises with TypeError: issubclass() arg 1 must be a class in uplink.converters.
Am I taking the problem on the correct angle or am I simply try to use the framework in the wrong way ?

[EvaSDK]

One thing I thought was to merge the two classes in one but what bothers me is that the fields are disjoint so I would have to make everything optional. Maybe https://pydantic-docs.helpmanual.io/usage/models/#generic-models is what I'm looking for.

[EvaSDK]

Does not look like a good solution in itself as one of the models has an enveloppe ("errors" key) while the other does not. So mapping from uplink directly does not seem possible this way.

[prkumar]

Unfortunately, this raises with TypeError: issubclass() arg 1 must be a class in uplink.converters. Am I taking the problem on the correct angle or am I simply try to use the framework in the wrong way ?

@EvaSDK - this may be related to #233, where our Pydantic converter does not handle Union properly.

[EvaSDK]

Interesting, I will have a look, thanks.

[EvaSDK]

I could finally get back to the code using this and I compiled the various solutions provided here.
Here is what I believe is a minimal solution for uplink users:

Edit: fixed support for unions with iterables.

import typing

from pydantic import BaseModel  # pylint: disable=no-name-in-module
from uplink.converters.pydantic_ import PydanticConverter as InitialConverter
from uplink.converters.pydantic_ import (
    _PydanticResponseBody as _InitialResponseBody,
)
from uplink.utils import is_subclass


class _PydanticResponseBody(_InitialResponseBody):
    """Pydantic converter supporting Union.

    See Also:
        :class:uplink.converters.pydantic._PydanticResponseBody
    """

    def __init__(self, model):
        super().__init__(model)
        self._union = False

        # Uplink does not natively support Union types
        # See https://github.com/prkumar/uplink/issues/233
        if typing.get_origin(self._model) is typing.Union:
            self._union = True

            class _UnionContainer(BaseModel):
                # pylint: disable=too-few-public-methods
                __root__: self._model

            self._model = _UnionContainer

    def convert(self, response):
        obj = super().convert(response)
        return obj.__root__ if self._union else obj


class PydanticConverter(InitialConverter):
    """Pydantic converter supporting Union types.

    Use this converter with :class:`uplink.Consumer` to enable support of union
    type annotations in HTTP route response.

    See Also:
        https://github.com/prkumar/uplink/issues/233
    """

    def _get_model(self, type_):
        if is_subclass(type_, BaseModel):
            return type_

        # Uplink does not natively support Union types
        # See https://github.com/prkumar/uplink/issues/233
        if typing.get_origin(type_) is typing.Union:
            if all(
                is_subclass(inner_type, BaseModel)
                or (
                    is_subclass(typing.get_origin(inner_type), Collection)
                    and all(
                        is_subclass(inner_type_child, BaseModel)
                        for inner_type_child in typing.get_args(inner_type)
                    )
                )
                for inner_type in typing.get_args(type_)
            ):
                return type_

        raise ValueError("Expected pydantic.BaseModel subclass or instance")

    def create_response_body_converter(self, type_, *args, **kwargs):
        """Build response body parser supporting Union types."""
        return self._make_converter(_PydanticResponseBody, type_)

Then it can be used like this:

github = Github(
    GITHUB_ENDPOINT,
    converter=PydanticConverter(),
)

[liiight]

I think that it can work if you set a Pydantic model using root: https://pydantic-docs.helpmanual.io/usage/models/#custom-root-types

something like:

class Success(BaseModel):
    code: conint(200, 299)
    message: str


class Failure(BaseModel):
    errors: Dict


class Response(BaseModel):
    __root__: Union[Success, Failure]

Didn't try this though...

[EvaSDK]

Unfortunately it fails like this:

In [13]: class Response(BaseModel):
    ...:     __root__ = Union[Success, Failure]
---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
Input In [13], in <module>
----> 1 class Response(BaseModel):
      2     __root__ = Union[Success, Failure]

File ~/Library/Virtualenvs/roc-tools/lib/python3.8/site-packages/pydantic/main.py:228, in pydantic.main.ModelMetaclass.__new__()

File ~/Library/Virtualenvs/roc-tools/lib/python3.8/site-packages/pydantic/fields.py:488, in pydantic.fields.ModelField.infer()

File ~/Library/Virtualenvs/roc-tools/lib/python3.8/site-packages/pydantic/fields.py:419, in pydantic.fields.ModelField.__init__()

File ~/Library/Virtualenvs/roc-tools/lib/python3.8/site-packages/pydantic/fields.py:539, in pydantic.fields.ModelField.prepare()

File ~/Library/Virtualenvs/roc-tools/lib/python3.8/site-packages/pydantic/fields.py:801, in pydantic.fields.ModelField.populate_validators()

File ~/Library/Virtualenvs/roc-tools/lib/python3.8/site-packages/pydantic/validators.py:723, in find_validators()

RuntimeError: no validator found for <class 'typing._GenericAlias'>, see `arbitrary_types_allowed` in Config

Adding arbitrary_types_allowed does not help:

In [17]: Response(__root__={"status": 200, "message": "OK"})
---------------------------------------------------------------------------
ValidationError                           Traceback (most recent call last)
Input In [17], in <module>
----> 1 Response(__root__={"status": 200, "message": "OK"})

File ~/Library/Virtualenvs/roc-tools/lib/python3.8/site-packages/pydantic/main.py:331, in pydantic.main.BaseModel.__init__()

ValidationError: 1 validation error for Response
__root__
  instance of _GenericAlias expected (type=type_error.arbitrary_type; expected_arbitrary_type=_GenericAlias)

[EvaSDK]

My current best result is:

class OperationResult(BaseModel):
    status: Optional[conint(ge=200, lt=600)]
    message: Optional[str]
    errors: Optional[Failure]

    @validator("errors", always=True)
    def check_success_or_error(cls, v, values):
        if v is not None and values.get("message") is not None:
            raise ValueError("Cannot be success and failure")
        if v is None and values.get("message") is None:
            raise ValueError("Must provide success or failure")
        return v

But if I need to subclass it (which happens in a couple of cases), I then have to make any extra fields optional as well so parsing does not fail in one case or the other.

[EvaSDK]

I merged your idea with that of @prkumar in #255 (reply in thread). Thanks for your help.