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, Generic

from fastapi import APIRouter, Depends
from pydantic.generics import GenericModel

# import your database base model here
# from db_model import BaseDBModel  
  
# your pydantic base model
# if you want to have get request params listed in doc,
# use model.generate_model_signature on your model
from model import CrudModel

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

class ListOf(GenericModel, Generic[T]):
    """Simple list model, feel free to put extra logic or fields in here"""
    
    count: int
    items: List[T]

    def __init__(self, items, **kwargs):
        super().__init__(items=items, count=len(items), **kwargs)


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=ListOf[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=ListOf[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 inspect import signature, Signature, Parameter
from itertools import islice
from typing import Type, Dict, Callable

from pydantic import BaseModel
from pydantic.main import MetaModel


def generate_model_signature(
    init: Callable[..., None], fields: Dict[str, 'ModelField'], config: Type['BaseConfig']
) -> Signature:
    """
    Generate signature for model based on its fields
    """

    present_params = signature(init).parameters.values()
    merged_params: Dict[str, Parameter] = {}
    var_kw = None
    use_var_kw = False

    for param in islice(present_params, 1, len(present_params)):  # skip self arg
        if param.kind is param.VAR_KEYWORD:
            var_kw = param
            continue
        merged_params[param.name] = param

    allow_names = config.allow_population_by_field_name
    for field_name, field in fields.items():
        param_name = field.alias
        if field_name in merged_params or param_name in merged_params:
            continue
        elif not param_name.isidentifier():
            if allow_names and field_name.isidentifier():
                param_name = field_name
            else:
                use_var_kw = True
                continue

        # TODO: replace annotation with actual expected types once #1055 solved
        kwargs = {'default': field.default} if not field.required else {}
        merged_params[param_name] = Parameter(param_name, Parameter.KEYWORD_ONLY, annotation=field.type_, **kwargs)

    if config.extra is config.extra.allow:
        use_var_kw = True

    if use_var_kw and var_kw is not None:
        merged_params[var_kw.name] = var_kw

    return Signature(parameters=list(merged_params.values()), return_annotation=None)


class CrudModelMeta(MetaModel):
    def __new__(mcs, name, bases, namespace):
        """
        Set real signature of created model, so FastAPI could see its real params
        Don't needed if this PR included in your version: https://github.com/samuelcolvin/pydantic/pull/1034
        """
        cls = super().__new__(mcs, name, bases, namespace)
        cls.__signature__ = generate_model_signature(cls.__init__, cls.__fields__, cls.__config__)
        return cls


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

    class Config:
        orm_mode = True