FastAPI 教程中 Pydantic 模型/模式之间的交互

Question

我遵循FastAPI 教程，但不太确定建议的数据对象之间的确切关系是什么。

我们有models.py文件：

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

和schemas.py文件：

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

然后，这些类用于定义数据库查询，如crud.py文件中所示：

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()

def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

在 FastAPI 代码中main.py：

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

据我了解：

• 数据models类定义 SQL 表。
• 数据schemas类定义 FastAPI 用于与数据库交互的 API。
• 它们必须可以相互转换，这样设置才能正常工作。

我不明白的是：

• 我crud.create_user_item预计返回类型为schemas.Item，因为 FastAPI 再次使用该返回类型。
• 根据我的理解@app.post("/users/{user_id}/items/", response_model=schemas.Item)中的响应模型main.py是错误的，或者我如何理解返回类型不一致？
• 但是从代码推断，实际的返回类型一定是models.Item，FastAPI 是如何处理的呢？
• 的返回类型是什么crud.get_user？

Answer 1

我将一一回顾您的要点。

数据models类定义 SQL 表。

是的。更准确地说，映射到实际数据库表的ormmodels类是在模块中定义的。

数据schemas类定义 FastAPI 用于与数据库交互的 API。

是和不是。模块中的 Pydantic 模型schemas定义了与 API 相关的数据模式，是的。但这与数据库无关。其中一些模式定义了某些 API 端点预期接收哪些数据，以使请求被视为有效。其他定义了某些端点返回的数据的样子。

它们必须可以相互转换，这样设置才能正常工作。

虽然数据库表模式和 API 数据模式通常非常相似，但情况并非一定如此。然而，在本教程中，它们的对应关系非常整齐，这允许简洁的CRUD代码，如下所示：

db_item = models.Item(**item.dict(), owner_id=user_id)

这是一个 Pydantic 模型实例，即您的 API 数据模式item之一，其中包含您认为创建新项目所需的数据。由于它的字段（它们的名称和类型）与数据库 model 的字段相对应，因此后者可以从前者的字典表示形式实例化（添加 ）。 schemas.ItemCreate models.Itemowner_id

我crud.create_user_item预计返回类型为schemas.Item，因为 FastAPI 再次使用该返回类型。

不，这正是FastAPI的魔力。该函数create_user_item返回 的实例models.Item，即从数据库构造的 ORM 对象（调用session.refresh它之后）：

def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    ...
    return db_item

API 路由处理程序函数 实际上create_item_for_user返回相同的对象（属于 类models.Item）。

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)

但是，装饰器会获取该对象并使用它来构造您为该路由定义@app.post的实例，在本例中就是如此。这就是您在模型中设置的原因：response_modelschemas.Itemorm_modeschemas.Item

class Config:
    orm_mode = True

这允许通过该方法创建该类的实例.from_orm。这一切都发生在幕后，并且再次取决于与 Pydantic 模型相对应的 SQLAlchemy 模型的字段名称和类型。否则验证失败。

根据我的理解，响应模型 [...] 是错误的

不，请参阅上文。修饰后的路由函数实际上返回模型的一个实例schemas.Item。

然而从代码推断，实际的返回类型必须是models.Item

是的，见上文。未修饰的路由处理函数的返回类型create_item_for_user实际上是models.Item。但它的返回类型不是响应模型。

我假设为了减少混乱，文档示例没有注释这些路由函数的返回类型。如果是的话，它看起来会像这样：

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
) -> models.Item:
    return crud.create_user_item(db=db, item=item, user_id=user_id)

记住函数装饰器只是一个函数的语法糖，它接受一个函数作为参数并（通常）返回一个函数，这可能会有所帮助。通常，返回的函数实际上在内部调用作为参数传递给它的函数，并在该调用之前和/或之后执行其他操作。我可以像这样重写上面的路线，它会完全相同：

def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
) -> models.Item:
    return crud.create_user_item(db=db, item=item, user_id=user_id)


create_item_for_user = app.post(
    "/users/{user_id}/items/", response_model=schemas.Item
)(create_item_for_user)

的返回类型是什么crud.get_user？

这是models.User因为这是数据库模型，也是first该查询方法返回的内容。

def get_user(db: Session, user_id: int) -> models.User:
    return db.query(models.User).filter(models.User.id == user_id).first()

read_user然后， API 路由函数再次以与我上面解释的相同的方式返回该值models.Item。

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)) -> models.User:
    db_user = crud.get_user(db, user_id=user_id)
    ...
    return db_user  # <-- instance of `models.User`

也就是说，该models.User对象被装饰器的内部函数拦截并（因为定义了response_model）传递给schemas.User.from_orm，它返回一个schemas.User对象。

希望这可以帮助。