️ SQLAlchemy 暴击指南把数据库变成 Python 对象直接拼 SQL 字符串又脏又容易注入纯手写 SQL 又重复又难维护。SQLAlchemy 把这个矛盾解开了——用 Python 类描述表用 Python 表达式写查询它替你生成正确的 SQL。这篇把 Obsidian 里 SQLAlchemy 的卡片和笔记揉成一条线从为什么需要 ORM一直到异步 事务每一段代码都能跑、版本都对齐。你来检查我兜底。 目录为什么需要 ORM / SQLAlchemy四大核心概念现代模型定义2.0 写法连库 建表完整 CRUD进阶查询关系映射事务要么全成要么全撤原生 SQL兜底手段异步 SQLAlchemy自检清单1. 为什么需要 ORM / SQLAlchemy先说人话ORM Object Relational Mapping对象关系映射。它把数据库表映射成Python 类把一行数据映射成一个对象。于是你不用写 SQL 字符串而是写 Python。SQLAlchemy 有两层初学者最容易混淆先分清层角色什么时候用Core核心偏底层的 SQL 表达能力表、语句、引擎要精细控制 SQL、或写原生查询时ORM对象关系把表映射成类用对象操作数据绝大多数业务代码本文重点# ❌ 裸 SQL 字符串sqlSELECT * FROM users \WHERE namename# 拼接 → 注入风险 难维护# ✅ SQLAlchemy ORMawaitsession.execute(select(User).where(User.namename))# 参数化、安全、可读一句话定位SQLAlchemy 是 Python 生态里事实标准的数据库工具包。FastAPI 官方教程用的就是它。学会它等于打通了Python 后端怎么存数据的任督二脉。2. 四大核心概念把下面四个词刻进脑子后面全是基于它们的组合概念是什么类比Engine数据库连接的总入口管理连接池水厂总管道Session一次和数据库对话的工作单元你和水厂的一次通话Base所有模型类的父类声明式基类表的图纸模板Model一张表对应一个类字段即列一张具体的表深挖连接池Engine不会每次查询都新建连接而是维护一个连接池——重复利用连接避免频繁握手开销。这也是为什么高并发下要用连接池而不是每次连一次。对应 Wiki 卡片《数据库连接池》。3. 现代模型定义2.0 写法这是全文最该记牢、也最容易踩版本坑的地方。SQLAlchemy 2.0 推荐使用DeclarativeBaseMappedmapped_column。# models.py · 2.0 风格fromsqlalchemyimportString,Integer,Floatfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnclassBase(DeclarativeBase):passclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50))age:Mapped[int]mapped_column(Integer,default18)score:Mapped[float]mapped_column(Float)版本坑必看否则过不了你的检查旧教程里常写的from sqlalchemy.ext.declarative import declarative_base再Base declarative_base()以及Column(Integer)那种写法在 2.0 里已废弃deprecated。新项目请一律用上面DeclarativeBaseMappedmapped_column的写法。遇到老代码能看懂即可自己写别再用旧的。类型怎么写Python 侧数据库列类型写法intINTEGERMapped[int] mapped_column(primary_keyTrue)strVARCHARMapped[str] mapped_column(String(50))floatFLOATMapped[float] mapped_column(Float)boolBOOLEANMapped[bool] mapped_column(defaultFalse)4. 连库 建表用create_engine建引擎再Base.metadata.create_all按模型建表开发/演示够用生产请用 Alembic 做迁移。# database.py · 同步fromsqlalchemyimportcreate_enginefrom.modelsimportBase# echoTrue 会把生成的 SQL 打到控制台学习期很有用enginecreate_engine(sqlite:///./demo.db,echoTrue)# 首次建表已存在则跳过Base.metadata.create_all(engine)连接串速查不同数据库只是连接串不同sqlite:///./x.db、postgresqlpsycopg://user:pwdlocalhost/db、mysqlpymysql://user:pwdlocalhost/db。换库基本只改这一行。5. 完整 CRUDCreate 增、Read 查、Update 改、Delete 删。下面是一套能直接跑的同步示例# crud.pyfromsqlalchemy.ormimportSessionfrom.databaseimportenginefrom.modelsimportUser# 增CreatewithSession(engine)ass:uUser(namexushuai,age20)s.add(u)s.commit()# 必须 commit 才真正写入s.refresh(u)# 把数据库生成的 id 同步回对象print(u.id)# 此时才有值# 查ReadwithSession(engine)ass:us.get(User,1)# 按主键查最快print(u.name)# 改UpdatewithSession(engine)ass:us.get(User,1)u.age21# 改属性即改记录s.commit()# 删DeletewithSession(engine)ass:us.get(User,1)s.delete(u)s.commit()⚠️最常见的两个坑①忘了commit()——内存里改了库里没动。②忘了refresh()就读 id——自增主键是数据库生成的commit 后还需 refresh 才能拿到。这两个点面试/实操高频出现。6. 进阶查询2.0 推荐用select()构造语句再用session.execute(...)执行scalars()取对象列表。fromsqlalchemyimportselect,func,or_# 条件过滤where 等价于旧版 filterstmtselect(User).where(User.age18)userss.scalars(stmt).all()# 或条件stmtselect(User).where(or_(User.age18,User.age60))# 模糊匹配LIKE %帅%stmtselect(User).where(User.name.contains(帅))# 或手写 likeUser.name.like(%帅%)# 排序 分页stmtselect(User).order_by(User.age.desc()).offset(0).limit(10)# 聚合总数 / 平均年龄totals.scalar(select(func.count()).select_from(User))avg_ages.scalar(select(func.avg(User.age)))Join 联表配合下一节的关系# 查出xushuai 写的所有文章stmtselect(Article).join(User).where(User.namexushuai)articless.scalars(stmt).all()7. 关系映射表与表之间有关系SQLAlchemy 用relationship()ForeignKey把它们变成对象间的引用。一对多一个用户写多篇文章fromsqlalchemyimportForeignKeyfromsqlalchemy.ormimportrelationship,Mapped,mapped_columnclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50))articles:Mapped[list[Article]]relationship(back_populatesauthor)classArticle(Base):__tablename__articlesid:Mapped[int]mapped_column(primary_keyTrue)title:Mapped[str]mapped_column(String(100))user_id:Mapped[int]mapped_column(ForeignKey(users.id))author:Mapped[User]relationship(back_populatesarticles)用起来就像操作对象user.articles直接拿到他的所有文章article.author直接拿到作者。一对一一个用户对应一份资料在一的那侧加uselistFalse即可classUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)profile:Mapped[Profile]relationship(back_populatesuser,uselistFalse)# 关键classProfile(Base):__tablename__profilesid:Mapped[int]mapped_column(primary_keyTrue)bio:Mapped[str]mapped_column(String(200))user_id:Mapped[int]mapped_column(ForeignKey(users.id))user:Mapped[User]relationship(back_populatesprofile)8. 事务要么全成要么全撤事务保证一组操作原子性要么全部成功提交要么出错整体回滚不会出现钱扣了但订单没生成的半吊子状态。withSession(engine)ass:try:s.add(User(nameA))s.add(User(nameB))s.commit()# 两条一起落库exceptException:s.rollback()# 出错 → 全部撤销库里干干净净raise小知识在 2.0 里with Session() as s:这个上下文管理器本身就有正常退出自动提交、异常退出自动回滚的能力。上面显式写try/except rollback是为了可读和可控也是面试里展示我懂事务的标准写法。9. 原生 SQL兜底手段ORM 覆盖 90% 场景但遇到复杂报表、窗口函数等直接写 SQL 更省心。用text()安全传参别直接拼字符串fromsqlalchemyimporttextwithengine.connect()asconn:resultconn.execute(text(SELECT * FROM users WHERE age :age),{age:18},# 参数化防注入)forrowinresult:print(row)# row 是类似元组的对象可按列名取10. 异步 SQLAlchemy高并发接口要用异步版create_async_engineasync_sessionmakerAsyncSession。注意数据库驱动也要换异步的如 PostgreSQL 用asyncpgSQLite 用aiosqlite。# database_async.py · 异步fromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromsqlalchemyimportselectfrom.modelsimportUser# 连接串前缀多了 aiosqlite / asyncpgenginecreate_async_engine(sqliteaiosqlite:///./demo.db)AsyncSessionLocalasync_sessionmaker(engine,expire_on_commitFalse)asyncdefget_users():asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(User))returnresult.scalars().all()和 FastAPI 配合时用lifespan在启动时建表用yield依赖把 Session 注入接口# main_async.py · FastAPI 集成fromcontextlibimportasynccontextmanagerfromfastapiimportFastAPI,DependsfromtypingimportAnnotatedasynccontextmanagerasyncdeflifespan(app:FastAPI):asyncwithengine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)# 启动建表yield# 应用运行期appFastAPI(lifespanlifespan)asyncdefget_db():asyncwithAsyncSessionLocal()assession:yieldsession# 注入后自动关闭app.get(/users/)asyncdeflist_users(db:Annotated[AsyncSession,Depends(get_db)]):resawaitdb.execute(select(User))returnres.scalars().all()深挖同步 vs 异步 怎么选学习/小项目用同步create_engineSession最省心。要扛高并发、配合async def接口才上异步。注意异步必须配异步驱动且 ORM 操作要await。两篇博客打通后你会发现FastAPI 管接口SQLAlchemy 管数据两者用 Depends 一接就活了。自检清单点开看答案Q1Engine、Session、Base、Model 四者分别是什么角色Engine连接总入口/连接池Session一次数据库对话的工作单元增删改查都在它里Base所有模型的声明式父类Model一张表对应一个类。关系Engine 造 SessionSession 操作 Model 实例Model 继承自 Base。Q2SQLAlchemy 2.0 里定义模型正确的写法是什么旧的 declarative_base() 还能用吗正确写法class Base(DeclarativeBase): pass字段用Mapped[类型] mapped_column(...)。旧declarative_base()Column()写法在 2.0 中已废弃能跑但不推荐新代码别用。Q3为什么 add 之后还要 commit有时还要 refreshcommit()才真正把改动写入数据库不 commit 只是内存里的挂起状态。refresh(obj)把数据库生成的值如自增id同步回 Python 对象之后才能真正拿到obj.id。Q4一对多和一对一在 relationship 上的区别是什么多的那一侧就是普通relationship如User.articles是列表一的那一侧加uselistFalse如User.profile是单个对象。两端用back_populates互指对方属性名保持双向同步。Q5事务的 rollback 解决什么问题什么时候该用它解决一组操作只成功了一部分的不一致问题。只要多个写操作必须要么全成、要么全撤如转账扣款入账就要放进同一个 Session出错时rollback()整体回滚保证数据原子性。Q6异步 SQLAlchemy 相比同步改了哪几处① 引擎换create_async_engine② 会话换async_sessionmakerAsyncSession③ 连接串加异步驱动前缀aiosqlite/asyncpg④ 所有 ORM 操作用awaitawait session.execute(...)。 资料来源 版本核查本篇整理自 Obsidian 知识库03 - 参考资料/数据库/03.SQLAlchemy学习与使用.md、FastAPI 第 07–09 章以及06 - Wiki/数据库/卡片SQLAlchemy Core / ORM / Session / Engine / 连接池 / Declarative Base。版本基准已联网核对2026-08-03SQLAlchemy 2.0.x最新稳定线 2.0.512.1 处于 2.1.0b3 beta未 GA。代码全部采用 2.0 现代写法。关键准确性说明declarative_base()自 2.0 起废弃改用DeclarativeBase2.0 推荐select()where()execute()/scalars()查询范式。