FastAPI：(11)SQL数据库

由于CSDN无法展示「渐构」的「#d，#e，#t，#c，#v，#a」标签，推荐访问我个人网站进行阅读：Hkini
「渐构展示」如下：在这里插入图片描述

#c 概述文章内容概括

1.安装SQLModel

#d SQLModel

SQLModel 是一个用于在 Python 中声明数据库模型的库，它结合了 SQLAlchemy 的 ORM 能力与 Pydantic 的数据验证特性。它旨在提供一种简单、统一的方式来定义数据库表结构，同时用于数据序列化、验证和交互，主要用于与 FastAPI 等现代 Web 框架集成。
安装SQLModel通过pip install sqlmodel

重要特征：

双重用途（ORM + 数据验证）：同时兼容 SQLAlchemy 的 ORM 和 Pydantic 的数据校验与序列化。
基于类型注解：使用 Python 的类型提示定义字段和数据结构，更直观地说明数据模型。
自动生成表结构：可通过模型类定义自动生成数据库表。
对异步和同步均友好：兼容异步和同步数据库操作，适应不同项目架构。
简化模型继承与组合：可继承基础类构建只读模型、创建数据模型、更新模型等多种用途。

#e 电子病历模型(正例) SQLModel

例子描述
医院系统中的电子病历模型 MedicalRecord，可表示为数据库表，也可用于接收医生上传记录或给前端返回。字段如 patient_id: int, diagnosis: str, record_time: datetime，模型用于数据库操作和前后端数据传输，完全符合 SQLModel 的理念。

特征对比

✅ 双重用途：模型同时作为数据库表结构与 API 的请求/响应模型。
✅ 基于类型注解：所有字段通过类型注解标明类型。
✅ 表结构生成：可通过该模型自动创建数据表。
✅ 异步友好：可用于 FastAPI 异步接口处理。

from sqlmodel import Field, SQLModel, create_engine, Session
from fastapi import FastAPI
from typing import Optional
from datetime import datetimeclass MedicalRecord(SQLModel, table=True):id: Optional[int] = Field(default=None, primary_key=True)patient_id: intdiagnosis: strrecord_time: datetimesqlite_url = "sqlite:///./test.db"
engine = create_engine(sqlite_url, echo=True)
SQLModel.metadata.create_all(engine)app = FastAPI()@app.post("/records/")
def create_record(record: MedicalRecord):with Session(engine) as session:session.add(record)session.commit()session.refresh(record)return record

#e 商品库存模型(正例) SQLModel

例子描述
电商平台中的 InventoryItem 模型记录商品 ID、库存数量、价格等信息。该模型用于库存管理模块的数据表定义，且用于与 API 交互中的库存变更和查询，统一使用。

特征对比

✅ 双重用途：模型用于数据库记录和接口的数据通信。
✅ 基于类型注解：如 quantity: int, price: float。
✅ 自动生成表结构：可以直接映射为 inventory_items 表。
✅ 简化模型继承：可以轻松派生创建更新库存的模型。

from sqlmodel import SQLModel, Field, create_engine, Session
from fastapi import FastAPI
from typing import Optionalclass InventoryItem(SQLModel, table=True):id: Optional[int] = Field(default=None, primary_key=True)name: strquantity: intprice: floatengine = create_engine("sqlite:///./inventory.db", echo=True)
SQLModel.metadata.create_all(engine)app = FastAPI()@app.post("/items/")
def add_item(item: InventoryItem):with Session(engine) as session:session.add(item)session.commit()session.refresh(item)return item

#e 外部API响应模型(反例) SQLModel

例子描述
一个天气服务中，用于解析外部 API 返回数据的 WeatherResponse 模型，字段如 temperature: float, humidity: float, location: str。该模型只用于临时解析 JSON 响应数据，不涉及数据库存储，也不参与任何数据写入操作。

特征对比

❌ 双重用途缺失：仅用于解析数据，不能表示数据库表。
❌ 缺乏 ORM 功能：字段虽有类型注解，但无数据库元数据。
❌ 无表结构生成意义：该数据模型不参与数据库操作。
✅ Pydantic 使用合理：仅作为数据验证使用，适合用 Pydantic 而非 SQLModel。

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optionalapp = FastAPI()# 外部天气服务返回的数据格式
class WeatherResponse(BaseModel):location: strtemperature: floathumidity: floatdescription: Optional[str]# 模拟调用外部 API，返回数据结构
@app.get("/weather/", response_model=WeatherResponse)
def get_weather():# 模拟外部 API 响应fake_api_data = {"location": "Chengdu","temperature": 28.5,"humidity": 70.2,"description": "Cloudy"}return fake_api_data

2.单一模型

#e 官网例子(正例) SQLModel

具体步骤：

创建模型
创建引擎
创建表
创建会话（Session）依赖项：Session 会存储内存中的对象并跟踪数据中所需更改的内容，然后它使用 engine 与数据库进行通信。使用 yield 创建一个 FastAPI 依赖项，为每个请求提供一个新的 Session 。这确保每个请求使用一个单独的会话。
启动创建表：对于生产环境，可能会用一个能够在启动应用程序之前运行的迁移脚本如Alembic。
创建Hero类：因为每个 SQLModel 模型同时也是一个 Pydantic 模型，所以可以在与 Pydantic 模型相同的类型注释中使用它。
读取Hero类
读取单个Hero
删除单个Hero

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, table=True): # 利用SQLModel创建数据库模型，table=True，告诉SQLModel这是一个表模型id: int | None = Field(default=None, primary_key=True) # primary_key表示是主键；int | None 数据在SQL应该的是INTEGER并且NULLABLE name: str = Field(index=True) # index=True 创建SQL索引，str在数据库中将会是TEXT或者VARCHARTage: int | None = Field(default=None, index=True)secret_name: strsqlite_file_name = "database.db" # 创建数据库引擎，用来与数据库保持连接
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False} # 不同线程中使用同一个SQLite数据库，将会按照代码结构确保「每个请求使用一个单独的SQLModel会话」
engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables(): # 创建表SQLModel.metadata.create_all(engine)def get_session(): # 创建Session会话依赖项with Session(engine) as session:yield session # 为每个请求提供一个新的SessionSessionDep = Annotated[Session, Depends(get_session)]app = FastAPI()@app.on_event("startup") # 启动时床啊金数据库表
def on_startup():create_db_and_tables()@app.post("/heroes/") # 创建Hero类，声明一个Hero参数，将从json主体中读取数据，同样可以声明为「返回类型」，自动生成API文档界面
def create_hero(hero: Hero, session: SessionDep) -> Hero:session.add(hero)session.commit()session.refresh(hero)return hero@app.get("/heroes/") # 读取「Hero」类，并利用limit和offset对结果进行分页
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes@app.get("/heroes/{hero_id}") # 读取单个「Hero」
def read_hero(hero_id: int, session: SessionDep) -> Hero:hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero@app.delete("/heroes/{hero_id}") # 删除单个「Hero」
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}

3.多个模型

#c 说明多个模型

现在稍微重构一下这个应用，以提高安全性和多功能性。

如果查看之前的应用程序，可以在 UI 界面中看到，到目前为止，由客户端决定要创建的 Hero 的 id 值。

不应该允许这样做，因为可能会覆盖在数据库中已经分配的 id 。决定 id 的行为应该由后端或数据库来完成，而非客户端。

此外，为 hero 创建了一个 secret_name ，但到目前为止，在各处都返回了它，这就不太秘密了……😅

通过添加一些额外的模型来解决这些问题，而 SQLModel 将在这里大放异彩。

#e 官网多模型例子(正例) SQLModel

操作步骤：

创建HeroBase基类
创建Hero表模型
创建HeroPublic公共数据模型
创建Hero的数据创建模型HeroCreate
创建Hero的更新模型HeroUpdate：HeroUpdate 数据模型有些特殊，它包含创建新 hero 所需的所有相同字段，但所有字段都是可选的（都有默认值）。这样，当更新一个 hero 时，可以只发送您想要更新的字段。因为所有字段实际上都发生了变化（类型现在包括 None ，并且现在有一个默认值 None ），需要重新声明它们，重新声明所有字段，因此并不是真的需要从 HeroBase 继承。让它继承只是为了保持一致，但这并不必要。这更多是个人喜好的问题。
使用HeroCreate创建并返回HeroPublic：在请求中接收到一个 HeroCreate 数据模型，然后从中创建一个 Hero 表模型。这个新的表模型 Hero 会包含客户端发送的字段，以及一个由数据库生成的 id 。然后将与函数中相同的表模型 Hero 原样返回。但是由于使用 HeroPublic 数据模型声明了 response_model ，FastAPI 会使用 HeroPublic 来验证和序列化数据。
用HeroPublic读取Hero类
用HeroPublic读取单个Hero
用HeroPublic更新单个Hero

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass HeroBase(SQLModel): # 创建HeroBase基类，「共享字段」name,agename: str = Field(index=True)age: int | None = Field(default=None, index=True)class Hero(HeroBase, table=True): # 创建Hero表模型，继承HeroBase的共享字段id: int | None = Field(default=None, primary_key=True)secret_name: strclass HeroPublic(HeroBase): # 创建返回给API客户端的HeroPublic模型，不包括secret_name，保护Heroid: int #　重新声明 `id: int` 。这样便与 API 客户端建立了一种约定，始终可以期待 `id` 存在并且是一个整数 `int`（永远不会是 `None` ）class HeroCreate(HeroBase):　# 创建用于创建hero的数据模型secret_name: str #不仅拥有与 `HeroBase` 相同的字段，还有 `secret_name` 。当客户端创建一个新的hero时，会送 `secret_name` ，被存储到数据库中，但这些 `secret_name` 不会通过 API 返回给客户端class HeroUpdate(HeroBase): # 创建用于更新Hero的数据模型name: str | None = Noneage: int | None = Nonesecret_name: str | None = Nonesqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()@app.post("/heroes/", response_model=HeroPublic) # 使用HeroCreate创建，并返回HeroPublic
def create_hero(hero: HeroCreate, session: SessionDep):db_hero = Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_hero # 将与函数中相同的表模型 `Hero` 原样返回。但是由于使用 `HeroPublic` 数据模型声明了 `response_model` ，**FastAPI** 会使用 `HeroPublic` 来验证和序列化数据。@app.get("/heroes/", response_model=list[HeroPublic]) # 用HeroPublic读取Hero表，使用 `response_model=list[HeroPublic]` 确保正确地验证和序列化数据。
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
):heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes@app.get("/heroes/{hero_id}", response_model=HeroPublic) #用HeroPublic读取单个Hero
def read_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero@app.patch("/heroes/{hero_id}", response_model=HeroPublic) #用`HeroUpdate` 更新单个 Hero
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db = session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code=404, detail="Hero not found")hero_data = hero.model_dump(exclude_unset=True) #在代码中，会得到一个 `dict` ，其中包含客户端发送的所有数据，只有客户端发送的数据，并排除了任何一个仅仅作为默认值存在的值。为此，用 `exclude_unset=True` 。这是最主要的技巧。hero_db.sqlmodel_update(hero_data) # 利用 `hero_data` 的数据更新 `hero_db`session.add(hero_db)session.commit()session.refresh(hero_db)return hero_db@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}