SQLAlchemy ORM 入门指南（Python）
SQLAlchemy 是 Python 的数据库工具与 ORM：模型定义、会话管理、增删改查、关系映射与查询表达式。本文用 SQLite 从零构建数据驱动应用。
直接回答：SQLAlchemy 是 Python 生态提供 Core 与 ORM 两层 API 的数据库工具：用 Python 类映射数据库表、用表达式构建查询、用会话管理事务——从玩具项目到大型系统都能驾驭，是独立数据库访问层的常见选择。
核心组件地图
SQLAlchemy 分两层：Core（SQL 表达式语言，贴近 SQL 本身）与 ORM（对象映射）。ORM 三要素：
- Engine：连接池与方言的入口；
- Session：工作单元——跟踪变更、批量提交、事务边界；
- Model：映射到表的 Python 类。
安装与模型定义
pip install "sqlalchemy>=2,<3"
from sqlalchemy import String, ForeignKey, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase): ...
class Author(Base):
__tablename__ = "authors"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
books: Mapped[list["Book"]] = relationship(back_populates="author")
class Book(Base):
__tablename__ = "books"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
author: Mapped[Author] = relationship(back_populates="books")
engine = create_engine("sqlite:///library.db")
Base.metadata.create_all(engine)
2.0 风格的 Mapped[] + mapped_column 让模型自带完整类型注解——可改进类型推断，但不能静态保证 SQL 对目标数据库有效。
新增
from sqlalchemy.orm import Session
with Session(engine) as session:
author = Author(name="王小波")
author.books = [Book(title="黄金时代"), Book(title="沉默的大多数")]
session.add(author) # 级联保存 books
session.commit()
Session 是工作单元：对象变更由 Session 跟踪；flush（也可能由查询自动触发）先向数据库发送 SQL，commit 再提交事务。flush 失败后若复用该 Session，需显式 rollback；上下文退出会关闭会话并回滚未完成事务。
查询
from sqlalchemy import select
with Session(engine) as session:
# 全表 / 条件 / 排序
books = session.scalars(select(Book).where(Book.title.contains("时代"))).all()
# 关联查询（避免 N+1）
from sqlalchemy.orm import selectinload
authors = session.scalars(
select(Author).options(selectinload(Author.books))
).all()
# 聚合
from sqlalchemy import func
cnt = session.scalar(select(func.count(Book.id)))
selectinload 是 N+1 问题的标准解药——关联数据批量预取。
更新与删除
with Session(engine) as session:
book = session.scalar(select(Book).where(Book.title == "黄金时代"))
if book is None:
raise LookupError("未找到目标图书")
book.title = "黄金时代（修订版）"
session.commit() # 脏检查自动发现变更
session.delete(book) # 删除
session.commit()
生产要点
- Session 生命周期：请求级创建销毁（Web 框架都有集成方案），不要跨线程或并发任务共享同一 Session；AsyncSession 也需每个并发任务独立使用；
- 连接池：
pool_size/max_overflow按并发调，池耗尽是生产事故常客； - 迁移：配 Alembic 管理表结构演进（
alembic revision --autogenerate），人工审查迁移后再执行；create_all 只用于示例初始化，不能替代生产迁移。
常见问题（FAQ）
Q：SQLAlchemy 和 Django ORM 怎么选？
A：Django 项目用内置 ORM；Flask/FastAPI/脚本等可按查询复杂度考虑 SQLAlchemy、轻量 ORM 或直接驱动，不能一概而论。2.0 的类型注解让体验差距大幅缩小。
Q：同步还是异步（AsyncSession）？
A：异步框架（FastAPI 等）且并发 IO 是瓶颈时用异步；传统同步应用用同步版，简单稳定。混用要非常小心边界。
Q：什么时候用 Core 而非 ORM？
A：批量数据搬运、动态 SQL、报表类复杂查询——Core 的表达式更贴近 SQL 且性能开销更小。日常业务 ORM，重型数据 Core，两者可以混用。
官方参考
本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。
