FastAPI + SQLAlchemy 进阶：多对多关系实战

大家好，上次我们学习了如何使用FastAPI和SQLAlchemy构建一对多的父子关系API。今天，让我们更进一步，探讨更复杂的多对多关系场景——学生选课系统。

为什么需要多对多关系？

在实际业务中，很多关系都不是简单的一对多：

学生和课程：一个学生可以选多门课，一门课也可以被多个学生选
文章和标签：一篇文章可以有多个标签，一个标签可以对应多篇文章
用户和角色：一个用户可以有多个角色，一个角色可以分配给多个用户

这就是典型的多对多关系。今天我们就以学生选课系统为例，深入剖析多对多关系的实现。

项目结构概览

我们将构建一个完整的选课系统API，包含以下功能：

学生管理（增删改查）
课程管理（增删改查）
学生选课/退课
查询学生及其所选课程

代码深度解析

1. 环境准备（同上篇）

1	pip install fastapi sqlalchemy uvicorn pydantic

2. 数据库配置（新增echo功能）

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, 
    connect_args={"check_same_thread": False},
    echo=True  # 新增：打印SQL语句，便于调试
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

关键点：

echo=True：开发时开启可以看到SQLAlchemy执行的SQL语句，对调试很有帮助

3. 关联表的设计（核心难点）

# 关联表定义（多对多的桥梁）
student_course = Table(
    "student_course",
    Base.metadata,
    Column("student_id", Integer, ForeignKey("students.id")),
    Column("course_id", Integer, ForeignKey("courses.id"))
)

关联表的作用：

它不对应业务实体，只记录学生和课程的关系
包含两个外键，分别指向学生表和课程表
联合主键确保不会重复选课

小贴士：在多对多关系中，这个中间表是最关键的设计，它把复杂的关系拆解成了两个一对多关系。

4. ORM模型设计（双向关系）

class StudentDB(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    # 多对多关系定义
    courses = relationship(
        "CourseDB", 
        secondary=student_course,  # 指定关联表
        back_populates="students"
    )

class CourseDB(Base):
    __tablename__ = "courses"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    students = relationship(
        "StudentDB", 
        secondary=student_course,  # 同样指定关联表
        back_populates="courses"
    )

relationship参数详解：

secondary：指定多对多关系的中间表
back_populates：建立双向关系，两边都可以访问
有了这个配置，我们可以：
- student.courses 获取学生的所有课程
- course.students 获取课程的所有学生

5. Pydantic模型

class Course(BaseModel):
    name: str

    class Config:
        orm_mode = True

class Student(BaseModel):
    name: str
    courses: list[Course] = []  # 嵌套显示课程列表

    class Config:
        orm_mode = True

这里我们复用了上篇的模式，但注意Student中包含courses列表，这会在查询时自动填充。

6. API端点实现

基础创建接口

@app.post("/students/", response_model=Student)
def create_student(student: Student, db: Session = Depends(get_db)):
    db_student = StudentDB(name=student.name)
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student

@app.post("/courses/", response_model=Course)
def create_course(course: Course, db: Session = Depends(get_db)):
    db_course = CourseDB(name=course.name)
    db.add(db_course)
    db.commit()
    db.refresh(db_course)
    return db_course

选课接口（多对多关系操作）

@app.post("/students/{student_id}/courses/{course_id}/")
def add_course_to_student(student_id: int, course_id: int, db: Session = Depends(get_db)):
    # 查询学生和课程
    student = db.query(StudentDB).filter(StudentDB.id == student_id).first()
    course = db.query(CourseDB).filter(CourseDB.id == course_id).first()
    
    # 验证存在性
    if student is None or course is None:
        raise HTTPException(status_code=404, detail="Student or Course not found")
    
    # 核心操作：将课程添加到学生的课程列表
    student.courses.append(course)
    db.commit()
    
    return {"message": "Course added to student successfully"}

神奇之处：student.courses.append(course) 这个简单的操作，SQLAlchemy会自动：

检查是否已存在关联
在中间表student_course中插入一条记录
处理所有数据库事务

查询学生及其课程

@app.get("/students/{student_id}/", response_model=Student)
def read_student(student_id: int, db: Session = Depends(get_db)):
    student = db.query(StudentDB).filter(StudentDB.id == student_id).first()
    if student is None:
        raise HTTPException(status_code=404, detail="Student not found")
    return student

查询时，由于我们在模型定义了relationship，student.courses会自动加载（懒加载），返回的JSON会包含完整的课程信息。

退课接口

@app.delete("/students/{student_id}/courses/{course_id}/")
def remove_course_from_student(student_id: int, course_id: int, db: Session = Depends(get_db)):
    student = db.query(StudentDB).filter(StudentDB.id == student_id).first()
    course = db.query(CourseDB).filter(CourseDB.id == course_id).first()
    
    if student is None or course is None:
        raise HTTPException(status_code=404, detail="Student or Course not found")
    
    # 从列表中移除
    student.courses.remove(course)
    db.commit()
    
    return {"message": "Course removed from student successfully"}

remove()方法同样会自动处理中间表的记录删除。

实战演练

启动服务：

1	uvicorn main:app --reload

API测试流程

创建学生

1 2	POST /students/ Body: {"name": "张三"}

创建课程

POST /courses/
Body: {"name": "Python编程"}
POST /courses/
Body: {"name": "数据库设计"}

学生选课

1 2	POST /students/1/courses/1/ POST /students/1/courses/2/

查询学生选课情况

1	GET /students/1/

返回示例：

{
    "name": "张三",
    "courses": [
        {"name": "Python编程"},
        {"name": "数据库设计"}
    ]
}

学生退课

1	DELETE /students/1/courses/1/

进阶优化

1. 防止重复选课

可以在Pydantic模型层面或业务逻辑层增加校验：

def add_course_to_student(student_id: int, course_id: int, db: Session = Depends(get_db)):
    student = db.query(StudentDB).filter(StudentDB.id == student_id).first()
    course = db.query(CourseDB).filter(CourseDB.id == course_id).first()
    
    # 检查是否已经选过这门课
    if course in student.courses:
        raise HTTPException(status_code=400, detail="Course already selected")
    
    student.courses.append(course)
    db.commit()
    return {"message": "Course added successfully"}

2. 添加时间戳

可以在关联表中添加额外字段，如选课时间：

# 如果需要存储额外信息，需要使用模型类而不是单纯的表
class StudentCourse(Base):
    __tablename__ = "student_course"
    student_id = Column(Integer, ForeignKey("students.id"), primary_key=True)
    course_id = Column(Integer, ForeignKey("courses.id"), primary_key=True)
    select_time = Column(DateTime, default=datetime.utcnow)
    score = Column(Integer, nullable=True)

3. 分页查询所有学生及其课程

@app.get("/students/")
def list_students(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    students = db.query(StudentDB).offset(skip).limit(limit).all()
    return students

4. 使用异步SQLAlchemy

对于高性能需求，可以考虑使用异步版本：

1	from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

常见问题与解决方案

Q1: 为什么我的关联表没有自动创建？

A: 确保在定义所有模型后调用了Base.metadata.create_all(bind=engine)

Q2: 查询时如何避免N+1问题？

A: 使用joinedload或selectinload预加载：

from sqlalchemy.orm import selectinload

student = db.query(StudentDB)\
    .options(selectinload(StudentDB.courses))\
    .filter(StudentDB.id == student_id)\
    .first()