(Discussion) Potential idea for making ORM RestAPI-friendly?

[synchronizing]

Hey, this is a random thought that I figured some opinions would be nice. This is an attempt solution at an annoyance of RestAPI frameworks being set-up so that DRY principles are constantly broken. This solution is proposed as an addon to Tortoise, but perhaps it shouldn't - unsure.

For the sake of example, take the following FastAPI route Python pseudocode using Pydantic, and Tortoise:

class PydanticChannel(PydanticModel):
    name: str

class TortoiseChannel(TortoiseModel):
    name = CharField(max_length=128, unique=True)

@route.post("/")
async def new_channel(schema):
    if await models.TortoiseChannel.exists(name=schema.name):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="A channel with this name already exists.",
        )

    return await models.TortoiseChannel.create(**schema.dict())

@route.get("/{name}")
async def get_channel(name):

    # We don't use 'exists' because we assume hit_rate > miss_rate.
    chnl = await models.TortoiseChannel.get_or_none(name=name)

    if not chnl:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Channel cannot be found.",
        )

    return chnl

@route.patch("/{name}")
async def patch(name, schema):

    # We don't use 'exists' because we assume hit_rate > miss_rate.
    chnl = await models.TortoiseChannel.get_or_none(name=name)

    if not chnl:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Channel cannot be found.",
        )
    elif await models.TortoiseChannel.exists(name=schema.name):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="A channel with this name already exists.",
        )

    await chnl.update_from_dict(channel.dict()).save()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

@route.delete('/{name}')
async def delete(name):
     chnl = await models.TortoiseChannel.get_or_none(name=name)

    if not chnl:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Channel cannot be found.",
        )

    await chnl.delete()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

As you can see in the example above there is a lot of repetitive code between the different routes - noticeably, error '404' (not found) and '409' (conflict). All of this stems from objects that either do not exists, or do exists; something the database naturally has to query to find out anyhow. Wondering if this can be figured out, I've come up with a solution that utilizes a 'generate_get_model` function that generates a 'get_model' function to optimize this process.

On second pass, however, it came to mind that this would be an interesting problem that Tortoise could solve out of the box to guarantee (unique) support for RESTful designs. Take the following example (not definite, but something I believe might be interesting):

from fastapi import status, HTTPException

class TortoiseChannel(TortoiseModel):
    name = CharField(max_length=128, unique=True)

    class Rest:

        class HTTP_404_NOT_FOUND:
            status = status.HTTP_404_NOT_FOUND
            info = "Channel could not be found."
            exception = tortoise.exceptions.DoesNotExist

        class HTTP_409_CONFLICT:
            status = status.HTTP_409_CONFLICT
            info = "Channel with this information already exists."
            exception = tortoise.exceptions.IntegrityError
            condition = lambda model: model.name != "admin" # Random conditional for example sake.

    class Config:
        rest_exception = HTTPException

With a model like this, a .rest() function could be added that can indicate which error to look-out for and raise a specific (in the case of FastAPI) HTTPException (.rest() takes *args). Therefore, the code above would become something like:

@route.post("/")
async def new_channel(schema):
    return await models.TortoiseChannel.create(**schema.dict()).rest(status.HTTP_409_CONFLICT)

@route.get("/{name}")
async def get_channel(name):
    return await models.TortoiseChannel.get(name=name).rest(status.HTTP_404_NOT_FOUND)

@route.patch("/{name}")
async def patch(name, schema):
    await models.TortoiseChannel.exists(name).rest(status.HTTP_404_NOT_FOUND)
    await models.TortoiseChannel.update_from_dict(*schema.dict()).rest=(status.HTTP_409_CONFLICT)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

@route.delete('/{name}')
async def delete(name):
    await models.TortoiseChannel.delete(name=name).rest=(status.HTTP_404_NOT_FOUND)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Therefore, .rest() servers almost like an middleware that checks for either an exception thrown, or a condition of the previous call. If you take the following:

class TortoiseChannel(TortoiseModel):
    name = CharField(max_length=128, unique=True)

    class Rest:

        class HTTP_404_NOT_FOUND:
            status = status.HTTP_404_NOT_FOUND
            info = "Channel could not be found."
            exception = tortoise.exceptions.DoesNotExist

        class HTTP_409_CONFLICT:
            status = status.HTTP_409_CONFLICT
            info = "Channel with this information already exists or is invalid."
            exception = tortoise.exceptions.IntegrityError
            condition = lambda model: model.name != "admin"

    class Config:
        rest_exception = HTTPException

@route.get("/{name}")
async def get_channel(name):
    return await models.TortoiseChannel.get(name=name).rest(status.HTTP_404_NOT_FOUND)

@route.post("/")
async def new_channel(schema):
    return await models.TortoiseChannel.create(**schema.dict()).rest(status.HTTP_409_CONFLICT)

This would be equivalent to:

async def catch_404(model):
    try:
        return await model
    except tortoise.exceptions.DoesNotExist:
         raise HTTPException(status=status.HTTP_404_NOT_FOUND, info="Channel could not be found.")

async def catch_409_and_conditional(model):
    condition  = lambda model: model.name != "admin"
    info = "Channel with this information already exists or is invalid."

    try:
        if condition(ret):
            raise HTTPException(status=status.HTTP_409_CONFLICT, info=info)
        return await model
    except tortoise.exceptions.IntegrityError:
         raise HTTPException(status=status.HTTP_409_CONFLICT, info=info)

@route.get("/{name}")
async def get_channel(name):
    return await catch_404(models.TortoiseChannel.get(name=name))

@route.post("/")
async def new_channel(schema):
    return await catch_409_and_conditional(models.TortoiseChannel.create(**schema.dict()))

Something along those lines, or a completely different idea/interface that helps with the DRY situation with RESTful APIs. If anyone has any suggestion or would like to share how they tried solving the DRY issue, that would be lovely.

.rest() could be separated into .exceptions() and .conditionals(), but hopefully the point is through. :)

[long2ice]

I think that's too heavy and should not do by tortoise. And tortoise is just a ORM and is protocol independent. But thanks your idea.

[synchronizing]

I think that's too heavy and should not do by tortoise.

Understandable criticism. I have no rebuttal.

And tortoise is just a ORM and is protocol independent.

Indeed - I suppose I should've generalized the thought before coming here. The interface is not so much a suggestion for implementation, but an example/conversation starter for some system/interface to better optimize ORM's/Tortoise in general (I've working with RestAPIs for a bit and it was naturally what I ran with). Something new & useful for any application that is depended on repetitive validations and querying.

In general, a 'middleware' system that allows some extensive (yet, easy-to-use) interface for adding "middle of the queryset" actions.

await models.Channel.filter(...).middleware(CustomMiddleware).first()

I do apologize for utilizing different examples, ideas, and implementations to explain what I mean - goes to show that this is simply a thought proposition and not a potential solution.

[alexpantyukhin]

What do you think about making the separated package (repo) some Tortoise-Rest, which is based in the tortoise.

[synchronizing]

What do you think about making the separated package (repo) some Tortoise-Rest, which is based in the tortoise.

I'm unsure how I would go about that with monkey-patching. The only feasible way I could think of is to do server-platform specific modifications. Here is what I'm currently using (inside of an organization module called mstoolbox that supports only FastAPI + Pydantic + Tortoise). Reading documentation will make sense of the utilities of each:

"""
Utilities that modify or update certain configurations relating to the Tortoise
ORM database framework.
"""
import asyncio
import sys
from functools import lru_cache
from typing import Callable, Dict, Iterable, Optional, Tuple, Type, TypedDict, Union

from fastapi import HTTPException, status
from toolbox import classproperty
from tortoise import models

from .pydantic import Model as PydanticModel

class hashit(dict):
    """Converts a normal dictionary to one that is hashable.

    Example:
        dict = {"hello": "world"}
        dict = hashit(dict)
        print(dict.__hash__)
    """

    def __hash__(self):
        return hash((frozenset(self), frozenset(self.values())))

class ErrorType(TypedDict):
    """Descriptive 'typing.Dict' type."""

    info: str
    exception: Optional[Exception]
    condition: Optional[Callable]

class Model(models.Model):
    """Patch that adds a few functionalities to the default Tortoise model."""

    @classproperty
    def fetch_fields(cls) -> Iterable:  # pylint: disable=no-self-argument
        """
        Adds a new class property called fetch_fields that includes a list of
        all relationships tied in with the model.

        This is intended to be used whenever there is a 'prefetch_related' or
        'fetch_related' function call that requires all of the fields to be
        fetched for. Removes the need for extra code.

        Example:
            model.prefetch_related(*model.fetch_fields)
            model.fetch_related(*model.fetch_fields)
        """
        return list(cls._meta.fetch_fields)

    @classmethod
    async def get_or_create(
        cls: Type[models.MODEL],
        *args,
        return_created=False,
        **kwargs,
    ) -> Union[
        models.MODEL, Tuple[models.MODEL, bool]
    ]:  # pylint: disable=arguments-differ
        """Gets the model passed. If it does not exist, creates it.

        This function is overwritten so that 'get_or_create' function is a bit
        more useful. In most cases we don't care whether or not the model was
        created, and rather simply just have it back. This overwrites the
        standard behavior of the 'get_or_create' function to give us the option
        to specify whether or not we want to know if the model existed or not
        before the function call.

        Args:
            cls: Tortoise model.
            return_created: Whether or not to return bool indicating if
                it was created.

        Example:
            model = Channel.get_or_create(name="grub")
            model, created = Channel.get_or_create(
                                name="grub",
                                return_created=True
                             )
        """
        model, created = await super(Model, cls).get_or_create(*args, **kwargs)

        if return_created:
            return model, created

        return model

    @classmethod
    def customize_get_or_none(
        cls: Type[models.MODEL],
        schema: PydanticModel,
        errors: Dict[int, ErrorType],
    ) -> None:
        """
        Replaces the model 'get_or_none' function with a custom one that allows
        running "middleware" like functionalities via exceptions and conditions.

        Support for FastAPI and Pydantic only.

        Info:
            The 'errors' argument needs to be specifically formatted as so:

            .. code-block:: python

                class ErrorType(TypedDict):
                    info: str
                    exception: Optional[Exception]
                    condition: Optional[Callable]

                errors: Dict[int, ErrorType]

            Every error must either have an 'exception' or a 'condition' - never
            both.

        Example:
            Say the following two models exist:

            .. code-block:: python

                class PydanticChannel(PydanticModel):
                    name: str

                class TortoiseChannel(TortoiseModel):
                    name = CharField(max_length=128, unique=True)

            Routes for 'Channel' must contain the usual ``GET``, ``POST``,
            ``PATCH``, and ``DELETE`` functions. The use of this function:

            .. code-block:: python

                TortoiseChannel.customize_get_or_none(
                    schema=PydanticChannel,
                    errors={
                        status.HTTP_404_NOT_FOUND: {
                            "info": "Could not locate this channel.",
                            "exception": tortoise.exceptions.DoesNotExist,
                        },
                        status.HTTP_409_CONFLICT: {
                            "info": "A channel with this information already exists.",
                            "condition": lambda model: bool(model),
                        },
                    },
                )

            Allows repetitive code to be stored away into a single interface
            that can be used for error and conditional checking. The 'model'
            and 'schema' arguments are trivial. The 'errors' argument indicates
            the possible HTTP error that may be throw depending on if
            'exception' or 'condition' are triggered.

            The 'exception' argument is caught when and if the original
            'Tortoise.Model.get' returns an error. The 'condition' is ran on
            the newly retrieved model from the database (assuming it did not
            throw any errors).

            Therefore, the basic HTTP functions for a resource can be more
            conveniently done like so:

            .. code-block::

                # Assuming we have access to the defined '_get_channel'
                # functionality above, and these functions serve as proper
                # routes.

                async def get(name):
                    return await TortoiseModel.get_or_none(name=name, errors=(status.HTTP_404_NOT_FOUND,))

                async def post(schema):
                    await TortoiseModel.get_or_none(schema=schema, errors=(status.HTTP_409_CONFLICT,))
                    return await models.ChannelModels.create(**channel.dict())

                asycn def delete(name):
                    chnl = await TortoiseModel.get_or_none(name=name, errors=(status.HTTP_404_NOT_FOUND,))
                    await chnl.delete()
                    return Response(status_code=status.HTTP_204_NO_CONTENT)

                async def patch(name, schema):
                    chnl = await TortoiseModel.get_or_none(name=name, errors=(status.HTTP_404_NOT_FOUND,))
                    await chnl.update_from_dict(schema.dict()).save()
                    return Response(status_code=status.HTTP_204_NO_CONTENT)

        Args:
            model: Tortoise model that will be used.
            schema: Pydantic schema to expect.
            errors: Formatted dictionary that contains status errors, exceptions,
                and conditionals.

        Returns:
            A callable that acts like 'Tortoise.Model.get_model' function.
        """

        # Stores parent function arguments into new variables due to same  naming
        # for the arguments used in the return function, 'get_model'. Documentation
        # refers to these as the 'DEFAULT' values.
        gen_model = cls
        gen_schema = schema
        gen_errors = errors

        # Ensures the errors list is formatted properly.
        for status, value in errors.items():
            if "info" not in value:
                err = "Error must have a detail message."
                raise Exception(err)

            if "exception" in value and "condition" in value:
                err = (
                    "Error must either have an 'exception' or 'condition'",
                    "function. Not both.",
                )
                raise Exception(err)

        @lru_cache
        def process_arguments(
            arguments: dict, errors: tuple, schema: bool
        ) -> Tuple[Dict, Dict, Dict]:
            """Process arguments given to 'get_model' function below.

            Args:
                schema: Pydantic schema.
                errors: Tuple of the errors given.
                **kwargs: Values that are valid entries to the DEFAULT schema.
            """

            # Ensures the PASSED errors are all part of the DEFAULT errors.
            #
            # This function also creates a 'current_errors' and
            # 'current_exceptions' dictionaries that are passed back to the calling
            # function to be used as optimization variables.
            #
            # More specifically, 'current_errors' is a subset of 'gen_errors' in
            # whose keys are the same set as the passed tuple 'error'.
            #
            # 'current_exceptions' is a subset of 'current_errors' that only
            # includes the errors that have exceptions (not conditionals), and is
            # formatted for quick retrieval of exceptions (exceptions being those
            # that inherit from the Exception object). In other words,
            #
            # current_errors = {400: {"detail": "Error.", "exception": DatabaseMissing}}
            # current_exceptions = {DatabaseMissing: {"status": 400, "detail": "Error."}}
            current_errors = {}
            current_exceptions = {}
            for error in errors:
                if error not in gen_errors.keys():
                    err = f"The passed '{error}' error was not defined in the original 'generate_query_function' function call."
                    raise Exception(err)

                # Generates the current_errors dictionary.
                current_errors[error] = gen_errors[error]

                # Generates the current_exceptions dictionary.
                value = gen_errors[error]
                if "exception" in value:
                    current_exceptions[value["exception"]] = {
                        "status": error,
                        "info": value["info"],
                    }

            # Grabs fields from the DEFAULT schema and model.
            schema_fields = tuple(gen_schema.__fields__.keys())
            model_fields = tuple(gen_model._meta.fields)

            # Verify PASSED schema values are all part of the DEFAULT model.
            if schema:
                for value in arguments.keys():
                    if value not in model_fields:
                        err = f"Input schema contains value '{value}' that is not part of the default model fields: {model_fields}"
                        raise Exception(err)

            # Verify PASSED kwargs values are all part of the DEFAULT schema.
            else:
                for karg in kwargs:
                    if karg not in schema_fields:
                        err = f"Input '{karg}' is invalid. The following inputs are valid: {schema_fields}"
                        raise Exception(err)

            return arguments, current_exceptions, current_errors

        async def _get_or_none(
            schema: PydanticModel = None,
            errors: tuple = (),
            **kwargs,
        ) -> PydanticModel:
            """Replacement function for 'get_model_or_none'.

            Args:
                schema: Pydantic schema.
                errors: Tuple of the errors given.
                **kwargs: Values that are valid entries to the DEFAULT schema.
            """

            # Throw error if both schema and kwargs are passed. User is forced to
            # use either/or. No error is thrown when neither are given and instead
            # function defaults to the normal 'get_model_or_none' function.
            if schema and kwargs:
                err = "'schema' and kwargs cannot be used together when calling this function."
                raise Exception(err)

            # Sets the proper querying arguments.
            if kwargs:
                arguments = kwargs
            elif schema:
                arguments = schema.dict()

            # Retrieves the relevant information to process the _get_model.
            arguments, current_exceptions, current_errors = process_arguments(
                arguments=hashit(arguments),
                errors=errors,
                schema=bool(schema),
            )

            # Tries querying the model.
            try:
                model = await gen_model.get(**arguments)
            except:
                # Grabs the specific exception that was thrown and verifies it with
                # the exceptions given.
                exception = sys.exc_info()[0]

                # Raises an HTTPException if the error was passed, otherwise sets
                # model to None.
                if exception in current_exceptions.keys():
                    value = current_exceptions[exception]
                    raise HTTPException(
                        status_code=value["status"],
                        detail=value["info"],
                    )
                else:
                    model = None

            # Runs conditionals on the model.
            for status, value in current_errors.items():
                if "condition" in value:
                    condition = value["condition"]

                    if condition(model):
                        raise HTTPException(
                            status_code=status,
                            detail=value["info"],
                        )

            # If all conditionals went through, return the queryset.
            return model

        cls.get_or_none = _get_or_none

[marcoaaguiar]

There is a package that implements CRUD nicely (even simpler than DRF) and it has integration with TortoiseORM (as well as other ORMs). I think this is what you wanted: https://github.com/awtkns/fastapi-crudrouter

(If so, this issue can be closed)

[synchronizing]

There is a package that implements CRUD nicely (even simpler than DRF) and it has integration with TortoiseORM (as well as other ORMs). I think this is what you wanted: https://github.com/awtkns/fastapi-crudrouter

(If so, this issue can be closed)

Bless your soul man, thank you for the share. Will definitely be diving deeper into this - just from the README, this is exactly what I was looking for. Valeu!