FastAPI CRUD fabric to reduce amount of boilerplate code

readme.md

...

class UserGet(CrudModel):
    id: int
    first_name: str
    last_name: str
    middle_name: str
    role: Optional[str]
    email: Optional[EmailStr]


class UserCreate(CrudModel):
    """
    Properties to receive on user create
    """
    first_name: str
    last_name: str
    middle_name: str
    email: Optional[EmailStr]


class UserUpdate(CrudModel):
    """
    Properties to receive on user update
    """
    first_name: Optional[str]
    last_name: Optional[str]
    middle_name: Optional[str]
    email: Optional[EmailStr]


class UserFilter(UserUpdate):
    role: Optional[str]
    email: Optional[EmailStr]

router = make_crud(
    User,
    name='user',
    create=UserCreate,
    get=UserGet,
    get_many=UserFilter,
    update=UserUpdate,
)

make_crud.py

import inspect
from functools import partial
from typing import Type, Literal, Callable, Dict, TypeVar, List

from fastapi import APIRouter, Depends

# import your database base model here
# from db_model import BaseDBModel
# this part is a bit outdated since you can now use SQLModel

# your pydantic base model
from model import CrudModel

T = TypeVar('T')
OnActionCallbacks = Dict[
    Literal['create', 'get', 'get_many', 'update', 'delete'],
    Callable[[T], T]
]


def path(param: str, annotation: T = inspect._empty) -> T:  # fake annotation
    """
    Dynamically creates path param with given name and type

    param = 'user_id'
    @router.get('{%s}' % param)
    def get_user(item_id=path(param, int)):
        assert type(item_id) is int
    """

    def param_dependency(**params):
        return params[param]

    # set fake signature with given name and annotation
    param_dependency.__signature__ = inspect.Signature(
        parameters=(
            inspect.Parameter(
                name=param,
                kind=inspect.Parameter.KEYWORD_ONLY,
                annotation=annotation
            ),
        )
    )
    return Depends(param_dependency)


def update_name(func: T, name: str) -> T:
    action = func.__name__
    func.__name__ = func.__qualname__ = action + name


def crud_actions(on_actions: , name):
    """
    Updates action name and wraps action with given callback
    """

    def decorator(func):
        action_name = func.__name__[:-1]
        update_name(func, name)
        if action_name in on_actions:
            action_wrapper = on_actions[action_name]
            return action_wrapper(func)
        return func

    return decorator


def make_crud(
        model: Type['BaseDBModel'],
        *,
        router: APIRouter = None,
        create: Type['CrudModel'] = None,
        get: Type['CrudModel'] = None,
        get_many: Type['CrudModel'] = None,
        update: Type['CrudModel'] = None,
        delete: Type['CrudModel'] = None,
        name: str = None,
        on: OnActionCallbacks = None
):
    if router is None:
        router = APIRouter()

    item_id = f"{name or 'item'}_id"
    crud_action = partial(
        crud_actions,
        on_actions=on or {},
        name=model.__tablename__
    )

    ...

    if create is not None:
        @router.post("/", response_model=get)
        @crud_action()
        async def create_(data: create) -> get:
            return model.create(data)

    ...

    if get is not None:
        @router.get('/{%s}' % item_id, response_model=get)
        @crud_action()
        async def get_(row_id=path(item_id, int)):
            return model.get(id=row_id)

    ...

    if isinstance(get_many, type):  # replace type with your CrudModelMeta
        @router.get("/", response_model=list[get], response_model_skip_defaults=True)
        @crud_action()
        async def get_many_(filters: get_many = Depends(get_many), skip: int = 0, limit: int = 100):
            return {'items': model.get_many(**filters.dict(exclude_none=True), skip=skip, limit=limit)}
          
    elif get_many:
        @router.get("/", response_model=list[get])
        @crud_action()
        async def get_many_(skip: int = 0, limit: int = 100):
            return {'items': model.get_many(skip=skip, limit=limit)}

    ...

    if update is not None:
        @router.put('/{%s}' % item_id, response_model=update, response_model_skip_defaults=True)
        @crud_action()
        async def update_(data: update, row_id=path(item_id, int)):
            return model.get(id=row_id).update(data)

    ...

    if delete is not None:
        @router.delete('/{%s}' % item_id)
        @crud_action()
        async def delete_(row_id=path(item_id)):
            model.where(id=row_id).delete()

    return router

model.py

from pydantic import BaseModel


class CrudModel(BaseModel):
    """
    A base config to use in models that are meant to help with ORM
    """

    class Config:
        orm_mode = True