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 生成候选迁移,审查后再执行。

官方参考

本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。

延伸阅读

获取专属方案

联系我们

加入社区

微信扫码
加入官方交流群

立即体验

在线开通,按量计费,真正的云服务!

立即开始

选择观测云版本

代码托管平台