FastAPI & SQLAlchemy 实战（九）：SQLAlchemy CRUD、关联关系与加载策略

「FastAPI & SQLAlchemy 实战」系列共 10 篇，从协程基础讲到数据库接口。这是第 9 篇。

结论：SQLAlchemy 的 CRUD 都围绕 Session：add + commit 新增，改属性 + commit 更新，delete + commit 删除，query/get 查询。关联关系用 relationship 配置。

一、新增

from datetime import date

def insert_data():
    db = SessionLocal()
    try:
        dept = Department(name="研发部", location="北京")
        db.add(dept)
        db.commit()
        db.refresh(dept)          # 取回自增 id

        db.add_all([
            Employee(name="张三", age=30, hire_date=date(2023, 1, 1), department_id=dept.id),
            Employee(name="李四", age=28, hire_date=date(2023, 3, 15), department_id=dept.id),
        ])
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

commit 后对象可能过期，refresh 用主键重新查一次，拿到自增 ID 和默认值。

二、删除与更新

# 删除
emp = db.query(Employee).filter(Employee.name == "李四").first()
if emp:
    db.delete(emp)
    db.commit()

# 更新
emp = db.query(Employee).filter(Employee.name == "张三").first()
if emp:
    emp.age = 31
    db.commit()

ORM 会自动跟踪对象属性的变化，commit 时生成 UPDATE；不需要手写 SQL。

三、查询

# 按主键
dept = db.get(Department, 1)

# 过滤
employees = db.query(Employee).filter(Employee.department_id == 1).all()

# and / or
from sqlalchemy import and_, or_
db.query(Employee).filter(and_(Employee.age.between(30, 40), Employee.department_id == 1))
db.query(Employee).filter(or_(Employee.department_id == 2, Employee.age > 32))

# join
result = db.query(Employee, Department).join(
    Department, Employee.department_id == Department.id
).all()

# 预加载，避免 N+1
from sqlalchemy.orm import joinedload
employees = db.query(Employee).options(joinedload(Employee.department)).all()

# 子查询：统计每个部门人数
from sqlalchemy import func
count_subq = db.query(
    Employee.department_id,
    func.count(Employee.id).label("count"),
).group_by(Employee.department_id).subquery()

# first / all
first = db.query(Employee).first()
all_depts = db.query(Department).all()

四、关联关系

# 一对多
class User(Base):
    items = relationship("Item", back_populates="owner", lazy="selectin")

class Item(Base):
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="items")

# 一对一：子表外键唯一，主表 uselist=False
user_id = Column(Integer, ForeignKey("users.id"), unique=True)
profile = relationship("Profile", back_populates="user", uselist=False)

多对多需要一张中间表：

student_course = Table(
    "student_course", Base.metadata,
    Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),
    Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True),
)

class Student(Base):
    courses = relationship("Course", secondary=student_course, back_populates="students")

五、加载策略

• lazy="select"（默认）：访问关联属性时才查询，容易 N+1；
• lazy="selectin"：用 IN 一次加载所有关联，推荐；
• lazy="joined"：JOIN 一起查，适合一对一；
• cascade="all, delete-orphan"：删除主对象时删关联对象，按业务选择。

六、常见错误

• 忘记 rollback：异常后会话不可用；
• 忘记 close：连接不归还；
• 循环里逐条查关联：N+1，用 selectin/joinedload；
• 乱用级联删除：删一个部门把员工全删了。

下一篇

最后一篇把 FastAPI 和 SQLAlchemy 接起来：用 Depends 注入 Session，写一组可运行的增删改查接口。

上一篇：SQLAlchemy 入门 ｜ 下一篇：FastAPI + SQLAlchemy 实战