带 SQLModel 的 FastAPI 响应模型¶

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

交互式 API 文档¶

到目前为止，我们使用的代码，API 文档知道客户端必须发送的数据

Interactive API docs UI

这个交互式文档 UI 由 Swagger UI 提供支持，Swagger UI 所做的是读取一个大的 JSON 内容，该内容使用标准 OpenAPI 定义 API 以及所有数据模式（数据形状），并将其显示在那个漂亮的 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+

# 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+

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

🤓 其他版本和变体

Python 3.9+Python 3.8+

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 列表。

为此，我们使用 list[Hero] 声明 response_model

Python 3.10+

# Code above omitted 👆

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

👀 完整文件预览

Python 3.10+

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

🤓 其他版本和变体

Python 3.9+Python 3.8+

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 和响应模型¶

FastAPI 将使用此 response_model 对响应进行数据验证和过滤。

因此，这就像我们的应用程序和客户端之间的一份契约。

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

新的 API 文档 UI¶

现在我们可以回到文档 UI，看到它们现在显示了我们将收到的响应模式。

API docs UI without response data schemas

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

自动客户端¶

使用 response_model 最明显的优点是它会显示在 API 文档 UI 中。

但还有其他优点，例如 FastAPI 将使用此模型对响应数据进行自动数据验证和过滤。

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

例如，客户端生成器，可以自动创建与你的 API 进行通信所需的多种语言代码。

信息

如果你对标准感到好奇，FastAPI 生成 OpenAPI，它内部使用 JSON Schema。

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

回顾¶

使用 response_model 告诉 FastAPI 你想要发送回的数据模式，并拥有出色的数据 API。😎

带 SQLModel 的 FastAPI 响应模型¶

交互式 API 文档¶

响应数据¶

使用 response_model¶

response_model 中的英雄列表¶

FastAPI 和响应模型¶

新的 API 文档 UI¶

自动客户端¶

回顾¶

使用 `response_model`¶

`response_model` 中的英雄列表¶