SQLAlchemy

基础概念

组成层次：

1. Engine：数据库连接与方言（dialect）管理
2. Connection / Session：事务性操作上下文
3. SQL Expression (Core)：可组合的 SQL 语法树
4. ORM：将类映射到表（Declarative）
5. Mapper：类与表之间映射配置

Core vs ORM：Core 偏底层显式构造 SQL；ORM 聚焦对象持久化与关系导航。

2.0 风格：去掉隐式自动提交 / 过时 API；推荐使用 with Session(engine) as session, session.begin()、select() 新式查询。

常见概念：

• 表达式构造：select() / insert() / update() / delete()
• 关系：relationship() + ForeignKey + back_populates
• 延迟加载策略：lazy / selectin / joined / subquery
• 会话生命周期：Session 代表一个工作单元（Unit of Work）
• 事务：Session.begin / commit / rollback
• Async 支持：sqlalchemy.ext.asyncio + async engine + async session

使用指南

安装

python -m pip install "SQLAlchemy>=2,<3"  # 同步 SQLite 示例无需额外驱动
# PostgreSQL 按实际 URL 驱动选择 psycopg2 或 asyncpg；不要无差别安装所有驱动。

定义模型（2.0 Declarative）

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, ForeignKey

class Base(DeclarativeBase):
  pass

class User(Base):
  __tablename__ = 'users'
  id: Mapped[int] = mapped_column(primary_key=True)
  name: Mapped[str] = mapped_column(String(50), index=True)
  articles: Mapped[list['Article']] = relationship(back_populates="author")

class Article(Base):
  __tablename__ = 'articles'
  id: Mapped[int] = mapped_column(primary_key=True)
  title: Mapped[str]
  author_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
  author: Mapped['User'] = relationship(back_populates="articles")

创建引擎与建表

from sqlalchemy import create_engine
engine = create_engine("sqlite:///./app.db", echo=False)
Base.metadata.create_all(engine)

以下同步片段按顺序运行，使用当前目录 app.db；重复运行会新增另一条 alice。临时演示可改为 sqlite+pysqlite:///:memory:。create_all 不负责升级已有表结构，生产变更使用版本迁移；应用结束时释放 Engine。

Session 使用

from sqlalchemy.orm import Session
with Session(engine) as session:
  u = User(name='alice')
  session.add(u)
  session.commit()

查询（2.0 select）

from sqlalchemy import select
with Session(engine) as session:
  stmt = select(User).where(User.name == 'alice')
  user = session.scalars(stmt).first()

关系加载策略

from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.articles))

异步示例

import os
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy import select

# PostgreSQL 示例另装 sqlalchemy[asyncio] 与 asyncpg；URL 来自运行环境。
async_engine = create_async_engine(os.environ["DATABASE_URL"], echo=False)
AsyncSessionFactory = async_sessionmaker(async_engine, expire_on_commit=False)

async def get_user(name: str):
    async with AsyncSessionFactory() as session:
        stmt = select(User.id, User.name).where(User.name == name)
        row = (await session.execute(stmt)).mappings().first()
        return dict(row) if row is not None else None

返回已物化的普通字段，不把依赖会话的 ORM 对象或未消费结果交给外层。name 在此模型不是唯一键，first 只取一行；需要唯一身份时按主键查询或建立唯一约束。应用关闭时等待 await async_engine.dispose()。AsyncSessionFactory 可复用，单个 AsyncSession 不供并发任务共享。

乐观锁

通过 mapper 的 version_id_col 与版本生成策略检测 ORM flush 更新冲突；版本字段通常设 NOT NULL。它不自动保护绕过逐对象 flush 的批量 UPDATE/DELETE，后者需显式版本条件并检查受影响行数。

Schema 演进

使用 Alembic：版本迁移（env.py + versions/ 生成脚本）。

常用模式

• Repository 抽象：封装常见 CRUD
• Unit of Work：统一提交/回滚（事务边界）
• DTO 分离：避免 ORM 模型直接暴露到 API

实战经验

1. 避免 n+1 查询：使用 selectinload / joinedload
2. 按工作单元创建 Session；一个请求可能有多个事务，不能跨线程或并发任务共享会话。
3. 提前规划索引：高频过滤列 + 组合索引
4. 2.x 批量插入可用 session.execute(insert(User), rows)；Core 用 engine.begin() 取得 Connection 再 conn.execute(…)。Engine.execute 已移除，bulk_save_objects 属于旧式 API。
5. commit 决定事务提交，flush 决定何时发 SQL；查询可能 autoflush，减少 commit 次数不保证减少 flush，事务也不宜无限延长。
6. 读写分离：自定义路由 Engine（读从写主）
7. 连接池调优：pool_size / max_overflow / pool_recycle
8. pool_timeout 是支持该参数的池等待连接超时，不是 SQL 执行超时；语句和锁超时需由数据库/驱动配置。
9. expire_on_commit=False 只控制提交后属性是否过期，不是对象内存限制；仍需管理结果规模与数据新鲜度。
10. 异步性能：I/O 密集时使用 async 版本 + asyncpg

常见坑：

• 忘记 Session 关闭导致连接泄漏
• 直接在模型属性里写复杂逻辑（可用 hybrid_property）
• 关系默认 lazy 导致隐式多次查询
• 大事务长时间占锁，需拆分
• ORM 模型与 Pydantic 模型耦合

经验总结

核心经验：清晰划分数据访问层，显式声明查询；对性能敏感部位回退 Core 手写 SQL；避免过度魔法。

注意：业务高并发下结合缓存 / CQRS / 读写分离 与 分页优化（覆盖索引）。

迁移规范：版本命名 + 严格 Review DDL 变更（锁表风险）。

信息参考

• 官方文档: https://docs.sqlalchemy.org
• 2.0 迁移指南: https://docs.sqlalchemy.org/en/20/changelog/migration_20.html
• ORM 加载策略: https://docs.sqlalchemy.org/en/20/orm/loading_relationships.html
• Async 文档: https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
• Alembic: https://alembic.sqlalchemy.org

Related notes

• Python MOC
• 数据访问层、事务与会话边界