SQLAlchemy 一对多关系完整指南：从建模到查询实战

在数据库设计中，一对多（One-to-Many）关系是最常见的关联关系之一。本文将通过一个完整的学校管理系统案例，详细介绍如何使用 SQLAlchemy ORM 实现一对多关系的建模、查询和操作。

一、场景介绍

我们将构建一个简单的学校管理系统，包含三层一对多关系：

• 班级（Class） ↔ 学生（Student）：一个班级有多个学生
• 学生（Student） ↔ 考试（Test）：一个学生有多次考试记录

这种层级结构在实际开发中非常常见，可以推广到订单-订单项、部门-员工等场景。

二、模型定义

2.1 基础模型类

首先定义一个包含时间戳的基类，所有模型都继承自这个基类：

import datetime
from sqlalchemy import Column, DateTime, JSON, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship

class Base(object):
    created_at = Column(
        DateTime,
        default=datetime.datetime.now,
        nullable=False
    )
    updated_at = Column(
        DateTime,
        default=datetime.datetime.now,
        onupdate=datetime.datetime.now,
        nullable=False
    )

DeclarativeBase = declarative_base(cls=Base)

2.2 班级模型（一方）

class Class(DeclarativeBase):
    __tablename__ = 'Class'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)  # 班级名称
    data = Column(JSON)    # JSON 字段存储额外信息

    # 关系定义：一个班级对应多个学生
    student_ = relationship('Student', back_populates='class_')

关键点：

• relationship 定义了与 Student 的关系
• back_populates 指定了反向引用的属性名

2.3 学生模型（多方）

class Student(DeclarativeBase):
    __tablename__ = 'Student'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)  # 学生名称

    # 外键：指向班级表
    class_id = Column(
        Integer,
        ForeignKey("Class.id", ondelete='CASCADE'),
        nullable=True,
        index=True
    )

    # 关系定义
    class_ = relationship(Class, back_populates='student_')
    test = relationship('Test', back_populates='student_')

关键点：

• ForeignKey 定义外键关联
• ondelete='CASCADE' 设置级联删除
• index=True 为外键创建索引，提高查询性能

2.4 考试模型（最底层多方）

class Test(DeclarativeBase):
    __tablename__ = 'Test'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)  # 考试名称

    student_id = Column(
        Integer,
        ForeignKey("Student.id", ondelete='CASCADE'),
        nullable=True,
        index=True
    )

    student_ = relationship(Student, back_populates='test')

三、外键约束说明

SQLAlchemy 支持三种外键删除策略：

策略	说明	使用场景
RESTRICT	有关联数据时拒绝删除（默认）	严格数据完整性
CASCADE	级联删除所有关联数据	需要自动清理关联数据
SET NULL	将外键设置为 NULL	保留子记录但解除关联

# CASCADE 示例
class_id = Column(Integer, ForeignKey("Class.id", ondelete='CASCADE'))

# SET NULL 示例（需要设置 nullable=True）
class_id = Column(Integer, ForeignKey("Class.id", ondelete='SET NULL'), nullable=True)

四、数据操作

4.1 创建数据库和会话

from sqlalchemy import create_engine
from sqlalchemy.orm import create_session

# 创建数据库引擎
sql = 'sqlite:///test2.db'
engine = create_engine(sql, echo=False)

# 创建所有表
DeclarativeBase.metadata.create_all(engine)

# 创建会话
session = create_session(bind=engine, autocommit=False, autoflush=True)

4.2 添加数据

# 创建班级
c = Class()
c.name = '高一1班'
c.data = {'age': 15, 'w': 3, 'dd': [33]}

# 创建学生并关联到班级
st = Student()
st.name = '黎明'
st.class_ = c  # 通过 relationship 直接关联

st2 = Student()
st2.name = '学友'
st2.class_ = c  # 同一个班级的另一个学生

# 创建考试记录并关联到学生
t = Test()
t.name = '期中考试'
t.student_ = st2

# 提交到数据库
session.add(c)
session.add(st)
session.add(st2)
session.add(t)
session.commit()

说明：

• 通过 relationship 定义的属性可以直接赋值对象
• SQLAlchemy 会自动处理外键关联
• 只需要 add 主对象，关联对象会自动保存

五、查询操作

5.1 基础查询

# 简单查询
dt = session.query(Class).filter(Class.id == 1).first()

# 多条件查询
from sqlalchemy import and_, or_

dt2 = session.query(Class).filter(
    and_(Class.id == 1, Class.name == '高一1班')
).first()

5.2 JSON 字段查询

# 查询 JSON 数组中的元素
d3 = session.query(Class).filter(
    Class.data['dd'][0].as_integer() > 5
).all()

for i in d3:
    print(i.data)

说明：

• SQLAlchemy 支持 JSON 字段的路径访问
• 使用 as_integer()、as_string() 等方法进行类型转换

5.3 正向查询（从多到一）

# 通过学生查询其所属班级
dt = session.query(Student).filter(Student.id == 1).first()
print(dt.class_)  # 直接访问 class_ 属性获取班级对象
print(f"学生 {dt.name} 所在班级：{dt.class_.name}")

优点：

• 不需要手动 JOIN
• 代码简洁直观
• 自动处理懒加载

5.4 反向查询（从一到多）

# 通过班级查询所有学生
dt = session.query(Class).filter(Class.id == 1).first()
print(dt.student_)  # 返回学生列表

# 遍历所有学生
for student in dt.student_:
    print(f"学生姓名：{student.name}")

5.5 跨级查询

# 从考试记录跨级查询到班级
dt = session.query(Test).filter(Test.id == 1).first()
print(dt.student_.class_)  # Test -> Student -> Class

# 输出考试对应的班级
print(f"考试 {dt.name} 对应的班级：{dt.student_.class_.name}")

六、更新和删除操作

6.1 更新数据

# 查询学生
dt = session.query(Student).filter(Student.id == 1).first()

# 修改属性
dt.name = '新名字'

# 提交修改
session.commit()

6.2 删除数据

# 删除班级（如果设置了 CASCADE，相关学生也会被删除）
dt = session.query(Class).filter(Class.id == 1).first()
session.delete(dt)
session.commit()

注意：

• 如果使用 RESTRICT 策略，删除有关联数据的记录会报错
• 使用 CASCADE 策略会自动删除所有关联的学生和考试记录
• 使用 SET NULL 策略会将学生的 class_id 设置为 NULL

七、会话状态管理

from sqlalchemy import inspect

# 查询对象
dt = session.query(Class).filter(Class.id == 1).first()

# 从会话中分离对象
session.expunge(dt)

# 检查对象状态
dd = inspect(dt)
print(dd.detached)  # True 表示对象已分离

# 分离所有对象
session.expunge_all()

使用场景：

• 跨线程传递对象
• 避免意外修改
• 序列化对象

八、完整示例代码

import datetime
from sqlalchemy import and_, Column, DateTime, JSON, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import declarative_base, relationship, create_session

# 基类定义
class Base(object):
    created_at = Column(DateTime, default=datetime.datetime.now, nullable=False)
    updated_at = Column(DateTime, default=datetime.datetime.now,
                       onupdate=datetime.datetime.now, nullable=False)

DeclarativeBase = declarative_base(cls=Base)

# 模型定义
class Class(DeclarativeBase):
    __tablename__ = 'Class'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    data = Column(JSON)
    student_ = relationship('Student', back_populates='class_')

class Student(DeclarativeBase):
    __tablename__ = 'Student'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    class_id = Column(Integer, ForeignKey("Class.id", ondelete='CASCADE'),
                     nullable=True, index=True)
    class_ = relationship(Class, back_populates='student_')
    test = relationship('Test', back_populates='student_')

class Test(DeclarativeBase):
    __tablename__ = 'Test'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    student_id = Column(Integer, ForeignKey("Student.id", ondelete='CASCADE'),
                       nullable=True, index=True)
    student_ = relationship(Student, back_populates='test')

# 创建数据库
engine = create_engine('sqlite:///test2.db', echo=False)
DeclarativeBase.metadata.create_all(engine)
session = create_session(bind=engine, autocommit=False, autoflush=True)

# 添加数据
c = Class(name='高一1班', data={'age': 15})
st = Student(name='黎明', class_=c)
st2 = Student(name='学友', class_=c)
t = Test(name='期中考试', student_=st2)

session.add_all([c, st, st2, t])
session.commit()

# 正向查询
student = session.query(Student).filter(Student.id == 1).first()
print(f"学生所在班级：{student.class_.name}")

# 反向查询
class_obj = session.query(Class).filter(Class.id == 1).first()
print(f"班级学生数量：{len(class_obj.student_)}")
for s in class_obj.student_:
    print(f"  - {s.name}")

session.close()

九、最佳实践

1. 外键索引：始终为外键字段创建索引（index=True）
2. 级联策略：根据业务需求选择合适的 ondelete 策略
3. 懒加载控制：对于大量数据，考虑使用 lazy='dynamic' 或 joinedload
4. 命名规范：使用清晰的命名区分表名和关系属性（如 class_ vs Class）
5. 会话管理：及时提交或回滚事务，避免会话泄漏

十、总结

本文通过一个完整的学校管理系统案例，展示了 SQLAlchemy 中一对多关系的核心用法：

• ✅ 使用 ForeignKey 定义外键
• ✅ 使用 relationship 建立双向关联
• ✅ 支持正向和反向查询
• ✅ 支持跨级关联查询
• ✅ 灵活的级联删除策略

掌握这些技巧后，你就能轻松应对各种复杂的数据库关系建模需求了。