logo标签列表

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

pythonsqlalchemyfastapipydantic
5
我按照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 再次使用了该返回类型。 - 根据我的理解，在 `main.py` 中 `@app.post("/users/{user_id}/items/", response_model=schemas.Item)` 的响应模型是错误的，或者我如何理解返回类型的不一致？ - 然而，从代码推断，实际的返回类型必须是 `models.Item`，FastAPI 如何处理这个问题？ - `crud.get_user` 的返回类型会是什么？
1个回答
14
我会逐一浏览你的要点。
数据类定义了SQL表格。 是的。更准确地说，映射到实际数据库表格的类在models模块中定义。

schemas数据类定义了FastAPI与数据库交互所使用的API。

是和不是。schemas模块中的Pydantic模型定义了与API相关的数据模式，没错。但这与数据库无关。其中一些模式定义了在特定API端点接收到请求时所期望的数据，以使请求被视为有效。其他模式则定义了某些端点返回的数据的结构。

它们必须可以相互转换，以使设置正常工作。 虽然数据库表模式和API数据模式通常非常相似，但并不一定如此。然而，在本教程中，它们相当整齐地对应，这允许使用简洁的代码，例如this： 
db_item = models.Item(**item.dict(), owner_id=user_id)
这里的item是一个Pydantic模型实例，即您的API数据模式之一schemas.ItemCreate，其中包含您决定用于创建新项目的必要数据。由于它的字段（名称和类型）对应于数据库模型models.Item的字段，因此可以从前者的字典表示中实例化后者（加上owner_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_model 的实例，本例中为 schemas.Item。这就是为什么您在 schemas.Item 模型中设置了 orm_mode 的原因。 
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)
可能有助于记住，function decorator只是一个语法糖，用于接受一个函数作为参数并（通常）返回一个函数。通常，返回的函数在调用传递给它的函数之前和/或之后执行其他操作。我可以像这样重写上面的路由，并且效果完全相同： 
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()
这个就像我上面解释models.Item一样，通过read_user API路由函数以同样的方式再次返回。 
@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对象。 希望这能帮到你。
3
非常感谢您提供这样清晰详尽的答案！ - patrick
1
这可能是FastAPI / Pydantic文档/教程的一部分。非常清晰！ - tomasyany
非常棒的回答！感谢 @Daniil Fajnberg，也感谢patrick的提问。 - undefined
网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，