SQLAlchemy 数据库实战指南

背景

数据库是大多数应用的持久化层。SQLAlchemy 让你既能在高级抽象上操作对象，又能随时下沉到原生 SQL 级别——它不是把你束缚在 ORM 里，而是给你一个从 Python 到 SQL 的完整工具箱。

---

第 1 章：Core vs ORM

# Core 层 — 直接 SQL 表达式
from sqlalchemy import text, select

with engine.connect() as conn:
    # 原生 SQL
    rows = conn.execute(text("SELECT * FROM users WHERE active = :active"), {"active": True})

    # SQL 表达式
    stmt = select(User).where(User.email.like("%@example.com"))
    result = conn.execute(stmt)


# ORM 层 — 对象操作
with Session(engine) as session:
    user = session.get(User, 1)
    user.name = "New Name"
    session.commit()  # 自动生成 UPDATE

---

第 2 章：关系加载策略

from sqlalchemy.orm import joinedload, selectinload, lazyload

# N+1 问题：默认 lazy loading，每次访问 post.author 发一条 SQL
# 解决方案 1：joinedload（JOIN 一次查出）
users = session.execute(
    select(User).options(joinedload(User.posts))
).unique().scalars().all()

# 解决方案 2：selectinload（IN 子查询，适合大数据量）
users = session.execute(
    select(User).options(selectinload(User.posts))
).unique().scalars().all()

---

第 3 章：事务与隔离级别

from sqlalchemy import select, update

# 显式事务
with Session(engine) as session:
    with session.begin():
        user = session.get(User, 1)
        user.balance -= 100
        session.add(
            TransactionLog(user_id=1, amount=-100, type="withdraw")
        )
    # 自动 commit，出错自动 rollback


# 悲观锁（SELECT ... FOR UPDATE）
stmt = select(User).where(User.id == 1).with_for_update()
user = session.execute(stmt).scalar_one()

---

第 4 章：Alembic 数据库迁移

# 初始化
alembic init alembic

# 生成迁移脚本
alembic revision --autogenerate -m "add users table"

# 应用迁移
alembic upgrade head

# 回滚
alembic downgrade -1

# 查看历史
alembic history

---

第 5 章：与 FastAPI 集成

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404)
    return {"id": user.id, "name": user.name}

---

思考题

1. joinedload 和 selectinload 各自的性能特征是什么？在什么情况下该选哪个？
2. SQLAlchemy Session 的 expire_on_commit 属性有什么作用？
3. AsyncSession + asyncpg 模式下，relationship 的 lazy loading 还能工作吗？