SQLModel 全面教程：常用 API 串联与实战指南

大家好，我是jobleap.cn的小九。
SQLModel 是一个专为 Python 设计的、融合了 Pydantic 和 SQLAlchemy 优势的 ORM（对象关系映射）库，它主要用来简化数据库操作与数据校验的流程，解决传统数据库开发中「数据模型定义重复」「类型校验繁琐」「SQL 编写复杂」等核心问题。

一、SQLModel 核心价值与解决的问题

1. 核心定位

SQLModel 由 FastAPI 作者 Sebastián Ramírez 开发，核心目标是：一份模型定义，同时满足数据库操作（SQLAlchemy）和数据校验（Pydantic）需求。

2. 解决的核心问题

• 传统开发中，需要分别定义 Pydantic 校验模型和 SQLAlchemy 数据库模型，存在大量重复代码；
• 手动编写 SQL 语句易出错，且缺乏类型提示，开发效率低；
• 数据库操作与数据校验的衔接繁琐，需要手动转换数据格式；
• 新手入门 ORM 时，SQLAlchemy 的复杂度较高，学习成本大。

二、环境准备

首先安装依赖（推荐使用虚拟环境）：

pipinstallsqlmodel python-dotenv# sqlmodel核心 + 环境变量管理# 如需连接MySQL/PostgreSQL，需额外安装驱动：# pip install pymysql # MySQL# pip install psycopg2-binary # PostgreSQL

三、核心 API 与基础使用

1. 定义数据模型（核心：单模型双用途）

SQLModel 的核心是SQLModel基类，继承它的类既可以作为数据库表模型，也可以作为 Pydantic 校验模型。

fromsqlmodelimportField,SQLModel,create_engine,Session,select# 1. 定义数据库模型（同时支持数据校验）classHero(SQLModel,table=True):"""英雄表模型：table=True 表示该类对应数据库表"""id:int|None=Field(default=None,primary_key=True)# 主键，自动递增name:str=Field(index=True)# 普通字段，建立索引age:int|None=Field(default=None,gt=0,lt=150)# 可选字段，校验年龄范围power:str|None=None# 可选字段，无额外约束# 2. 定义仅用于校验的模型（非数据库表）classHeroCreate(SQLModel):"""创建英雄时的入参校验模型（无需主键，由数据库自动生成）"""name:strage:int|None=Field(default=None,gt=0,lt=150)power:str|None=None

2. 数据库连接与表创建

# 1. 配置数据库连接（SQLite 无需额外驱动，适合快速测试）DATABASE_URL="sqlite:///./heroes.db"# 注意：SQLite 需要 check_same_thread=False（其他数据库无需）engine=create_engine(DATABASE_URL,connect_args={"check_same_thread":False})# 2. 创建所有表（仅需执行一次）defcreate_db_and_tables():SQLModel.metadata.create_all(engine)# 初始化数据库表if__name__=="__main__":create_db_and_tables()

3. 核心 CRUD API 实战

CRUD（创建/读取/更新/删除）是数据库操作的核心，以下是 SQLModel 常用 API 的串联实战：

（1）创建数据（Create）

defcreate_hero(hero:HeroCreate):# 将校验模型转换为数据库模型（自动忽略无关字段）db_hero=Hero.model_validate(hero)# 使用 Session 管理数据库连接withSession(engine)assession:session.add(db_hero)# 添加数据到会话session.commit()# 提交事务session.refresh(db_hero)# 刷新数据（获取自动生成的主键）returndb_hero# 调用示例if__name__=="__main__":create_db_and_tables()# 创建英雄iron_man=create_hero(HeroCreate(name="钢铁侠",age=45,power="高科技战甲"))spider_man=create_hero(HeroCreate(name="蜘蛛侠",age=18,power="蜘蛛感应"))print("创建的英雄：",iron_man,spider_man,sep="\n")

（2）读取数据（Read）

支持单条查询、列表查询、条件查询，核心 API 是select()+session.exec()：

defget_hero_by_id(hero_id:int)->Hero|None:"""根据ID查询单个英雄"""withSession(engine)assession:hero=session.get(Hero,hero_id)# 快捷查询主键returnherodefget_all_heroes(skip:int=0,limit:int=10)->list[Hero]:"""查询英雄列表（分页）"""withSession(engine)assession:statement=select(Hero).offset(skip).limit(limit)# 构建查询语句heroes=session.exec(statement).all()# 执行查询并获取所有结果returnheroesdefget_heroes_by_age(age:int)->list[Hero]:"""条件查询：根据年龄查询英雄"""withSession(engine)assession:statement=select(Hero).where(Hero.age==age)# 条件过滤heroes=session.exec(statement).all()returnheroes# 调用示例if__name__=="__main__":# 查询单个英雄hero_1=get_hero_by_id(1)print("ID=1 的英雄：",hero_1)# 查询所有英雄all_heroes=get_all_heroes()print("所有英雄：",all_heroes,sep="\n")# 条件查询young_heroes=get_heroes_by_age(18)print("18岁的英雄：",young_heroes)

（3）更新数据（Update）

defupdate_hero(hero_id:int,hero:HeroCreate)->Hero|None:"""更新英雄信息"""withSession(engine)assession:db_hero=session.get(Hero,hero_id)ifnotdb_hero:returnNone# 将新数据更新到数据库模型（仅更新传入的字段）hero_data=hero.model_dump(exclude_unset=True)# 排除未设置的字段forkey,valueinhero_data.items():setattr(db_hero,key,value)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero# 调用示例if__name__=="__main__":# 更新蜘蛛侠的年龄updated_spider=update_hero(2,HeroCreate(name="蜘蛛侠",age=19,power="蜘蛛感应+纳米战衣"))print("更新后的蜘蛛侠：",updated_spider)

（4）删除数据（Delete）

defdelete_hero(hero_id:int)->bool:"""删除英雄"""withSession(engine)assession:hero=session.get(Hero,hero_id)ifnothero:returnFalsesession.delete(hero)session.commit()returnTrue# 调用示例if__name__=="__main__":# 删除ID=1的英雄is_deleted=delete_hero(1)print("删除结果：","成功"ifis_deletedelse"失败")

4. 进阶用法：关联模型（一对一/一对多）

SQLModel 支持表关联，以下是「英雄-团队」一对多关联示例：

fromsqlmodelimportRelationship# 定义团队模型classTeam(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)# 关联英雄（反向引用：hero.team）heroes:list["Hero"]=Relationship(back_populates="team")# 扩展英雄模型，添加团队外键classHero(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)name:str=Field(index=True)age:int|None=Field(default=None,gt=0,lt=150)power:str|None=Noneteam_id:int|None=Field(default=None,foreign_key="team.id")# 外键team:Team|None=Relationship(back_populates="heroes")# 关联团队# 实战：创建团队并关联英雄if__name__=="__main__":create_db_and_tables()# 创建团队avengers=Team(name="复仇者联盟")withSession(engine)assession:session.add(avengers)session.commit()session.refresh(avengers)# 创建英雄并关联团队captain_america=Hero(name="美国队长",age=100,power="超级血清",team_id=avengers.id)session.add(captain_america)session.commit()# 查询团队及关联的英雄statement=select(Team).where(Team.name=="复仇者联盟")team=session.exec(statement).first()print("团队名称：",team.name)print("团队英雄：",team.heroes)

四、与 FastAPI 集成（实战延伸）

SQLModel 与 FastAPI 天然兼容（同作者），以下是快速集成示例：

fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModel app=FastAPI()# 初始化数据库@app.on_event("startup")defon_startup():create_db_and_tables()# 接口：创建英雄@app.post("/heroes/",response_model=Hero)defcreate_hero_api(hero:HeroCreate):returncreate_hero(hero)# 接口：查询单个英雄@app.get("/heroes/{hero_id}",response_model=Hero)defget_hero_api(hero_id:int):hero=get_hero_by_id(hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero# 运行：uvicorn main:app --reload

总结

1. 核心价值：SQLModel 实现「一份模型=数据校验+数据库操作」，消除重复代码，提升开发效率；
2. 核心 API：SQLModel（模型基类）、create_engine（数据库连接）、Session（会话管理）、select()（查询构建）、model_validate()/model_dump()（数据转换）是高频使用的核心 API；
3. 核心场景：适合快速开发 Python 后端项目（尤其是 FastAPI），简化数据库 CRUD 操作，同时保证数据校验的严谨性。

通过以上教程，你可以快速掌握 SQLModel 的核心用法，解决传统数据库开发中「模型重复」「校验繁琐」「SQL 易出错」等问题。