SQLModel 入门指南:SQLAlchemy 与 Pydantic 的模型协作
SQLModel 把 SQLAlchemy 的 ORM 与 Pydantic 的数据校验合进一个模型:共享公共字段,并分别定义数据库、请求和响应模型。本文讲解安装、模型、CRUD 与 FastAPI 集成。
直接回答:SQLModel 是 FastAPI 作者出品的库:共享基类可减少数据库与 API 模型间的字段重复,但请求、表、响应仍应按用途分离,避免将内部字段自动暴露给客户端。
解决的问题
FastAPI 项目的经典重复:User(SQLAlchemy 模型)、UserCreate(Pydantic 输入)、UserResponse(Pydantic 输出)——三个类描述同一份数据,改个字段改三处。SQLModel 的思路:一个基类承载共有字段,表模式与 API 模式按需派生。
安装与模型
pip install sqlmodel
from sqlmodel import SQLModel, Field, create_engine, Session, select
class Hero(SQLModel, table=True): # table=True = 映射到表
id: int | None = Field(default=None, primary_key=True)
name: str
power: str
age: int | None = None # 可空字段
engine = create_engine("sqlite:///heroes.db")
SQLModel.metadata.create_all(engine)
不要把 table=True 模型的普通构造过程当作请求输入的完整校验;使用独立输入模型,并通过 model_validate 等显式校验路径转换。table=True 注册表元数据,不会仅因定义类就自动修改数据库;不带 table 的基类用于共享数据字段,这套"共享基类"模式是 SQLModel 的精髓。
新增
with Session(engine) as session:
hero = Hero(name="蜘蛛侠", power="蛛丝", age=18)
session.add(hero)
session.commit()
session.refresh(hero) # 拿回自增 id
print(hero.id)
查询
with Session(engine) as session:
# 全部
heroes = session.exec(select(Hero)).all()
# 条件 + 排序 + 分页
stmt = select(Hero).where(Hero.age > 20).order_by(Hero.name).limit(10)
results = session.exec(stmt).all()
# 单条
hero = session.get(Hero, 1)
这里从 sqlmodel 导入的 select 包装了 SQLAlchemy 查询构造能力;SQLModel 查询就是 SQLAlchemy 查询,所有进阶能力(JOIN、聚合、预加载)原样可用。
更新与删除
with Session(engine) as session:
hero = session.get(Hero, 1)
if hero is None:
raise LookupError("未找到目标英雄")
hero.age = 19
session.commit()
session.delete(hero)
session.commit()
FastAPI 集成:共享基类模式
此节是替换前面模型的独立示例,不要在同一元数据中重复定义 Hero 表。同步 FastAPI 路由使用 SQLite 时,engine 需增加 connect_args={"check_same_thread": False};每个请求仍独立创建 Session,不能跨线程共享同一个 Session。
from fastapi import FastAPI, Depends
from sqlmodel import SQLModel, Field, Session
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
class HeroBase(SQLModel):
name: str
power: str
class Hero(HeroBase, table=True):
id: int | None = Field(default=None, primary_key=True)
class HeroCreate(HeroBase): ... # 输入模型
class HeroPublic(HeroBase): # 输出模型(带 id)
id: int
@app.post("/heroes", response_model=HeroPublic)
def create(hero: HeroCreate, session: Session = Depends(get_session)):
db_hero = Hero.model_validate(hero)
session.add(db_hero); session.commit(); session.refresh(db_hero)
return db_hero
基类定义一次,三个角色各取所需——公共字段可统一维护;敏感字段、默认值和可写范围仍需分别设计。表变更要执行迁移,旧客户端兼容性也需验证,不能理解为接口与数据库自动安全同步。
常见问题(FAQ)
Q:SQLModel 会取代 SQLAlchemy 吗?
A:不会——它建在 SQLAlchemy 之上,底层引擎、会话、查询都是 SQLAlchemy 的。它是"FastAPI 场景的最佳拍档",不是替代品。
Q:什么时候不该用 SQLModel?
A:复杂领域模型(需要精细控制映射)、非 API 项目(纯数据管道)、Django 项目。SQLModel 的甜区是 API 驱动的 CRUD 应用。
Q:迁移怎么做?
A:照样用 Alembic——SQLModel 的表元数据就是 SQLAlchemy 的,设置 target_metadata 后,用 alembic revision --autogenerate 生成候选迁移,审查后再执行。
官方参考
本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。