Search code examples
pythonsqlalchemyfastapipython-typingpydantic

Properly typed factory for FastAPI and Pydantic


I have been developing my first API using FastAPI/SQLAlchemy. I have been using the same four methods (Get One, Get All, Post, Delete) for multiple different entities in the database, thus creating a lot of repeated code. For example, the code below shows the methods for a Fungus entity.

from typing import List, TYPE_CHECKING

if TYPE_CHECKING: from sqlalchemy.orm import Session

import models.fungus as models
import schemas.fungus as schemas

async def create_fungus(fungus: schemas.CreateFungus, db: "Session") -> schemas.Fungus:
    fungus = models.Fungus(**fungus.dict())
    db.add(fungus)
    db.commit()
    db.refresh(fungus)
    return schemas.Fungus.from_orm(fungus)

async def get_all_fungi(db: "Session") -> List[schemas.Fungus]:
    fungi = db.query(models.Fungus).limit(25).all()
    return [schemas.Fungus.from_orm(fungus) for fungus in fungi]


async def get_fungus(fungus_id: str, db: "Session") -> schemas.Fungus:
    fungus = db.query(models.Fungus).filter(models.Fungus.internal_id == fungus_id).first()
    return fungus


async def delete_fungus(fungus_id: str, db: "Session") -> int:
    num_rows = db.query(models.Fungus).filter_by(id=fungus_id).delete()
    db.commit()
    return num_rows

I have been trying to come up with an abstract design pattern with an interface class that implements these four methods independent from the entity.

However, from my understanding new Python standards and FastAPI require Python to be typed. So, how would I type the return types of these functions, instead of schemas.Fungus, or the parameters schemas.CreateFungus or models.Fungus?

What I have thought is that I could use the type of these values, which are <class 'pydantic.main.ModelMetaclass'> and <class 'sqlalchemy.orm.decl_api.DeclarativeMeta'>. However I am not sure if this is correct or encouraged.


Solution

  • I am not sure why you want to go the abstract route. But a generic interface for those common CRUD operations is certainly possible.

    You could write your own generic class that is (for example) parameterized by

    1. the ORM model (from SQLAlchemy) to use for database transactions,
    2. the "response" model (from Pydantic) for things like getting/listing objects, and
    3. the "create" model (from Pydantic) for adding new data.

    Here is an example of such a class that is based on what you did with the fungi models:

    from typing import Generic, TypeVar
    
    from pydantic import BaseModel
    from sqlalchemy.orm import DeclarativeMeta, Session
    
    GetModel = TypeVar("GetModel", bound=BaseModel)
    CreateModel = TypeVar("CreateModel", bound=BaseModel)
    ORM = TypeVar("ORM", bound=DeclarativeMeta)
    
    
    class CRUDInterface(Generic[ORM, GetModel, CreateModel]):
        orm_model: ORM
        get_model: type[GetModel]
        create_model: type[CreateModel]
    
        def __init__(
            self,
            orm_model: ORM,
            get_model: type[GetModel],
            create_model: type[CreateModel],
            db_id_field: str = "id",
        ) -> None:
            self.orm_model = orm_model
            self.get_model = get_model
            self.create_model = create_model
            self.db_id_field = db_id_field
    
        def create(self, data: CreateModel, db: Session) -> GetModel:
            new_instance = self.orm_model(**data.dict())
            db.add(new_instance)
            db.commit()
            db.refresh(new_instance)
            return self.get_model.from_orm(new_instance)
    
        def list(self, db: Session, limit: int = 25) -> list[GetModel]:
            objects = db.query(self.orm_model).limit(limit).all()
            return [self.get_model.from_orm(obj) for obj in objects]
    
        def get(self, id_: int, db: Session) -> GetModel:
            where = getattr(self.orm_model, self.db_id_field) == id_
            obj = db.query(self.orm_model).filter(where).first()
            return self.get_model.from_orm(obj)
    
        def delete(self, id_: int, db: Session) -> int:
            filter_kwargs = {self.db_id_field: id_}
            num_rows = db.query(self.orm_model).filter_by(**filter_kwargs).delete()
            db.commit()
            return num_rows
    

    Now this class is fully generic in terms of the models involved in the CRUD operations you showed in your example, which means upon instantiation, an object of CRUDInterface will be properly typed in its methods, and as a bonus there is no need to pass around the specific models after initialization. The models are simply saved as instance attributes.

    Assume now that you have the following models: (simplified)

    from pydantic import BaseModel
    from sqlalchemy import Column, Integer, String
    from sqlalchemy.orm import DeclarativeMeta, declarative_base
    
    
    ORMBase: DeclarativeMeta = declarative_base()
    
    
    class Fungus(BaseModel):
        id: int
        name: str
    
        class Config:
            orm_mode = True
    
    
    class CreateFungus(BaseModel):
        name: str
    
    
    class FungusORM(ORMBase):
        __tablename__ = "fungus"
        id = Column(Integer, primary_key=True)
        name = Column(String)
    

    Here is a little usage demo with those models:

    from typing_extensions import reveal_type
    
    from sqlalchemy import create_engine
    from sqlalchemy.orm import Session
    
    # ... import CRUDInterface, ORMBase, FungusORM, Fungus, CreateFungus
    
    
    def main() -> None:
        engine = create_engine("sqlite:///", echo=True)
        ORMBase.metadata.create_all(engine)
        fungus_crud = CRUDInterface(FungusORM, Fungus, CreateFungus)
        with Session(engine) as db:
            mushroom = fungus_crud.create(CreateFungus(name="oyster mushroom"), db)
            reveal_type(mushroom)
            assert mushroom.id == 1
    
            same_mushroom = fungus_crud.get(1, db)
            reveal_type(same_mushroom)
            assert mushroom == same_mushroom
    
            all_mushrooms = fungus_crud.list(db)
            reveal_type(all_mushrooms)
            assert all_mushrooms[0] == mushroom
    
            num_deleted = fungus_crud.delete(1, db)
            assert num_deleted == 1
            all_mushrooms = fungus_crud.list(db)
            assert len(all_mushrooms) == 0
    
    
    if __name__ == "__main__":
        main()
    

    This executes without error and the generated SQL queries are as we would expect. Moreover, running mypy --strict over this code yields no errors (at least with the SQLAlchemy Mypy plugin) and reveals the types as expected, too:

    note: Revealed type is "Fungus"
    note: Revealed type is "Fungus"
    note: Revealed type is "builtins.list[Fungus]"
    

    This clearly demonstrates the advantage of using generics over simply annotating with the common base classes or meta classes.

    If you just annotated, say CRUDInterface.get with pydantic.BaseModel, a type checker would never be able to tell you what specific model instance will be returned. In practice, this means your IDE will not give you any suggestions about specific methods/attributes of e.g. the Fungus model.

    Obviously, there is a lot more you can do with generics in this context. You can also make the existing methods more flexible and so on. But this example should at least get you started.