FastAPI Response Model with SQLModel¶

现在我将向您展示如何将 FastAPI 的 response_model 与 SQLModel 一起使用。

交互式 API 文档¶

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

Interactive API docs UI

这个交互式文档 UI 由 Swagger UI 提供支持，Swagger UI 所做的是读取一个大型 JSON 内容，该内容使用标准 OpenAPI 定义了包含所有数据模式（数据形状）的 API，并在漂亮的 UI 中显示它。

FastAPI 自动为 Swagger UI 读取而生成 OpenAPI。

它基于您编写的代码生成它，使用 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` 中的 Hero 列表¶

我们还可以使用其他类型注解，就像我们可以使用 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 和 Response Model¶

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

因此，这就像我们的应用程序和客户端之间的合同。

您可以在关于 response_model 的 FastAPI 文档中阅读更多相关信息。

新的 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。 😎

FastAPI Response Model with SQLModel¶

交互式 API 文档¶

响应数据¶

使用 response_model¶

response_model 中的 Hero 列表¶

FastAPI 和 Response Model¶

新的 API 文档 UI¶

自动客户端¶

回顾¶

使用 `response_model`¶

`response_model` 中的 Hero 列表¶