FastAPI 响应模型与 SQLModel¶

现在，我将向你展示如何使用 FastAPI 的 response_model 与 SQLModel 配合使用。

交互式 API 文档¶

到目前为止，我们使用的代码，API 文档已经知道客户端需要发送的数据：

Interactive API docs UI

这个交互式文档 UI 是由 Swagger UI 提供支持的，Swagger UI 的作用是读取一个大型的 JSON 内容，这个 JSON 定义了 API 和所有数据架构（数据结构），使用标准的 OpenAPI，并将其显示在这个漂亮的 UI 中。

FastAPI 会自动 生成 OpenAPI，供 Swagger UI 读取。

它根据你编写的代码生成这些内容，利用 Pydantic 模型（在本例中是 SQLModel 模型）和类型注解来了解 API 处理的数据架构。

响应数据¶

但是，到目前为止，API 文档 UI 并不知道我们的应用程序返回的响应的架构。

你可以看到，有一个可能的 "成功响应" 和代码 200，但是我们并不知道响应数据的具体样子。

API docs UI without response data schemas

现在，我们只告诉了 FastAPI 我们希望接收的数据，但还没有告诉它我们希望发送回的数据。

让我们现在做这个改变。🤓

使用 `response_model`¶

我们可以使用 response_model 来告诉 FastAPI 我们希望发送回的数据的架构。

例如，我们可以传递相同的 Hero SQLModel 类（因为它也是一个 Pydantic 模型）：

Python 3.10+Python 3.9+Python 3.7+

# Code above omitted 👆

@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero

# Code below omitted 👇

# Code above omitted 👆

@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero

# Code below omitted 👇

# Code above omitted 👆

@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero

# Code below omitted 👇

👀 完整文件预览

Python 3.10+Python 3.9+Python 3.7+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

在 `response_model` 中列出英雄¶

我们还可以使用其他类型注解，就像我们在 Pydantic 字段中使用的那样。例如，我们可以传递一个 Hero 对象的列表。

首先，我们从 typing 导入 List，然后在 response_model 中声明 List[Hero]：

Python 3.10+Python 3.9+Python 3.7+

# Code here omitted 👈

@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

# Code below omitted 👇

# Code here omitted 👈

@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

# Code below omitted 👇

from typing import List, Optional

# Code here omitted 👈

@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

# Code below omitted 👇

👀 完整文件预览

Python 3.10+Python 3.9+Python 3.7+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=Hero)
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

FastAPI 和 Response Model¶

FastAPI 会使用这个 response_model 进行数据验证和响应数据过滤。

这就像是我们应用程序和客户端之间的契约。

你可以在 FastAPI 文档关于 response_model 的部分中阅读更多内容。

新的 API 文档 UI¶

现在我们可以返回到文档 UI，看到它们已经显示了我们将接收到的响应的 schema。

API docs UI without response data schemas

客户端将知道他们应该期待什么数据。

自动化客户端¶

使用 response_model 的最大好处之一就是它会显示在 API 文档 UI 中。

但它还有其他优点，比如 FastAPI 会使用这个模型自动进行数据验证和响应数据过滤。

此外，由于 schema 是使用标准定义的，有许多工具可以利用这一点。

例如，客户端生成器可以自动生成与您的 API 交互所需的代码，支持多种语言。

Tip

如果你对这些标准感兴趣，FastAPI 会生成 OpenAPI，而 OpenAPI 内部使用的是 JSON Schema。

你可以在 FastAPI 文档 - 第一步中阅读所有相关内容。

总结¶

使用 response_model 告诉 FastAPI 你希望发送回的数据的 schema，从而拥有强大的数据 API。 😎

FastAPI 响应模型与 SQLModel¶

交互式 API 文档¶

响应数据¶

使用 response_model¶

在 response_model 中列出英雄¶

FastAPI 和 Response Model¶

新的 API 文档 UI¶

自动化客户端¶

总结¶

使用 `response_model`¶

在 `response_model` 中列出英雄¶