Python FastAPI 和 Pydantic 的正确类型工厂

我不知道你为什么要走abstract路.但对于这些常见的CRUD操作，generic接口当然是可能的.

您可以编写您自己的generic class，它(例如)被参数化为

1. 用于数据库事务的ORM模型(来自SQLAlChemy)，
2. 获取/列出对象之类的东西的"响应"模型(来自Pydatics)，以及
3. 用于添加新数据的"创建"模型(来自Pydtic).

下面是这样一个类的示例，它基于您对真菌模型所做的操作:

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

现在，就您在示例中显示的CRUD操作所涉及的模型而言，这个类是完全泛型的，这意味着在实例化时，将在其方法中正确键入一个CRUDInterface的对象，并且不需要在初始化之后传递specific个模型.模型只是作为实例属性保存.

现在假设您有以下模型:(简化)

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)

以下是这些型号的一个小用法演示:

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()

它的执行没有错误，生成的SQL查询与我们预期的一样.此外，在此代码上运行mypy --strict不会产生任何错误(至少对于SQLAlchemy Mypy plugin)，并且也会如预期的那样显示types:

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

这清楚地展示了使用泛型相对于简单地使用公共基类或元类进行注释的优势.

如果只是用pydantic.BaseModel注释CRUDInterface.get，类型判断器永远不能告诉您将返回哪个specific模型实例.实际上，这意味着您的IDE不会为您提供有关Fungus模型的specific个方法/属性的任何建议.

显然，在这种情况下，您可以对泛型做更多的工作.您还可以使现有方法更加灵活，等等.但这个例子至少应该能让你开始学习.