持久化存储: SQL 数据库

持久内存：SQL 数据库

到目前为止，我们的 API 是无状态的。如果您重启服务器，添加到列表或字典中的任何数据都将消失。在现实世界中，应用程序需要持久化。在本章中，我们将使用 SQLModel 将 FastAPI 应用程序连接到真实的 SQL 数据库。

为什么选择 SQLModel？

FastAPI 适用于任何数据库（SQL 或 NoSQL）。然而，SQLModel 是由 FastAPI 的同一作者 (Tiangolo) 构建的库。它旨在成为 Pydantic（数据验证）和 SQLAlchemy（数据库交互）之间的完美桥梁。使用 SQLModel，您可以定义一个单一的类，它既可以作为 API 验证的 Pydantic 模型，也可以作为数据库表的 SQLAlchemy 模型。

基于模型创建表

首先，我们需要安装该库 (pip install sqlmodel)。然后，我们定义数据结构。通过继承 SQLModel 并传递 table=True，我们告诉该库这个类代表数据库中的一个表。

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine

# 这个类既作为 Pydantic 模型，也作为数据库表
class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None

# 创建数据库引擎（本例使用 SQLite）
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

# 创建表的辅助函数
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

通过依赖注入建立会话

要与数据库交互，我们需要一个 Session。我们希望每个请求都有一个全新的会话，并确保在请求完成后关闭它。这正是 FastAPI 结合 yield 进行依赖注入的完美用例。

from sqlmodel import Session
from fastapi import Depends, FastAPI

app = FastAPI()

# 依赖项
def get_session():
    with Session(engine) as session:
        yield session
        # 会话在此处（请求完成后）自动关闭

执行 CRUD 操作

现在我们可以在路径操作中直接实现 CRUD（创建、读取、更新、删除）。

创建 (POST)

我们注入 Session 依赖项，添加对象，提交事务，并刷新对象以获取生成的 ID。

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

读取 (GET)

我们可以使用 select 语句来查询数据。SQLModel 足够智能，可以将数据库结果转换回标准的 Python 对象。

from sqlmodel import select

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

通过将数据库会话作为依赖项集成，您的代码保持整洁、可测试，并能有效防止资源泄漏。