FastAPI + Tortoise ORM 课程选修管理系统——后端实现思路全解

一、项目概述

本文详细解析一个基于 FastAPI + Tortoise ORM + MySQL 8 的课程选修管理系统后端实现，涵盖从数据库建模、JWT 认证、ORM 关系映射到复杂统计查询的完整思路。适合正在学习 FastAPI 异步开发、Tortoise ORM 关系处理的后端开发者参考。

技术栈

组件	选型	说明
Web 框架	FastAPI	原生异步支持，自动生成 OpenAPI 文档
ORM	Tortoise ORM	Python 异步 ORM，Django 风格 API
数据库	MySQL 8	关系型数据库，支持事务
认证	python-jose + passlib	JWT Token + bcrypt 密码哈希
数据校验	Pydantic v2	FastAPI 内置，声明式校验

---

二、架构设计

分层架构

┌─────────────────────────────────────────┐
│  routers/   ← 路由层：接收请求、参数校验   │
├─────────────────────────────────────────┤
│  schemas.py ← 数据层：Pydantic 请求/响应   │
├─────────────────────────────────────────┤
│  models.py  ← 模型层：Tortoise ORM 映射    │
├─────────────────────────────────────────┤
│  auth.py    ← 认证层：JWT + 密码哈希       │
├─────────────────────────────────────────┤
│  config.py  ← 配置层：数据库/JWT 参数      │
└─────────────────────────────────────────┘

关键设计原则：

• 所有路由函数使用 async def，充分利用 FastAPI 异步能力
• 认证逻辑通过 Depends() 依赖注入，与业务路由解耦
• Pydantic Schema 分离请求体校验和响应体序列化，遵循关注点分离

---

三、数据库模型设计

3.1 ER 关系图（文字描述）

User ──→ Student（一对一，可选）
              │
              ├──→ StudentProfile（一对一，profile 反向关系）
              │
              └──→ Enrollment（多对多中间表，enrollments 反向关系）
                       │
                       └──→ Course（多对一，enrollments 反向关系）
                                │
                                └──→ Teacher（多对一，courses 反向关系）

3.2 六张核心表

class User(Model):
    """用户表 - 用于系统登录认证（管理员和学生均可登录）"""
    id = fields.BigIntField(pk=True)                                            # 主键，自增 ID
    username = fields.CharField(max_length=50, unique=True)                     # 用户名，全局唯一
    password = fields.CharField(max_length=255)                                 # 密码哈希（使用 passlib 加密存储）
    role = fields.CharField(max_length=20, default="student")                   # 角色：admin（管理员）/ student（学生）
    student = fields.OneToOneField("models.Student", null=True,                 # 一对一关联 Student（学生用户时绑定）
                                   related_name="user")                         # 反向关系名
    created_at = fields.DatetimeField(auto_now_add=True)                        # 创建时间，自动填充

    class Meta:
        table = "users"                                                         # 指定数据库表名

    def __str__(self):
        return f"User({self.username}, {self.role})"                            # 字符串表示


class Student(Model):
    """学生表 - 存储学生基本信息"""
    id = fields.BigIntField(pk=True)                                            # 主键，自增 ID
    student_no = fields.CharField(max_length=20, unique=True)                   # 学号，全局唯一
    name = fields.CharField(max_length=50)                                      # 学生姓名
    gender = fields.SmallIntField(default=0)                                    # 性别：0=未知，1=男，2=女
    enrollment_year = fields.IntField()                                         # 入学年份
    created_at = fields.DatetimeField(auto_now_add=True)                        # 创建时间，自动填充
    # 一对一反向关系：通过 profile 可访问 StudentProfile 对象
    profile: fields.ReverseRelation["StudentProfile"]                           # 类型标注，IDE 友好
    # 多对多反向关系：通过 enrollments 可访问该学生的所有选课记录
    courses: fields.ReverseRelation["Enrollment"]                               # 类型标注，IDE 友好

    class Meta:
        table = "students"                                                      # 指定数据库表名

    def __str__(self):
        return f"Student({self.student_no}, {self.name})"                       # 字符串表示


class StudentProfile(Model):
    """学生档案表 - 与学生一对一关联，存储详细信息"""
    id = fields.BigIntField(pk=True)                                            # 主键，自增 ID
    student = fields.OneToOneField("models.Student",                            # 一对一关联 Student
                                   related_name="profile")                      # 反向关系名（student.profile）
    id_card = fields.CharField(max_length=18, null=True)                        # 身份证号码
    phone = fields.CharField(max_length=20, null=True)                          # 手机号码
    email = fields.CharField(max_length=100, null=True)                         # 电子邮箱
    hometown = fields.CharField(max_length=100, null=True)                      # 籍贯
    emergency_contact = fields.CharField(max_length=50, null=True)              # 紧急联系人姓名
    emergency_phone = fields.CharField(max_length=20, null=True)                # 紧急联系人电话

    class Meta:
        table = "student_profiles"                                              # 指定数据库表名

    def __str__(self):
        return f"Profile({self.student_id})"                                    # 字符串表示


class Teacher(Model):
    """教师表 - 存储教师基本信息"""
    id = fields.BigIntField(pk=True)                                            # 主键，自增 ID
    name = fields.CharField(max_length=50)                                      # 教师姓名
    employee_no = fields.CharField(max_length=20, unique=True)                  # 工号，全局唯一
    department = fields.CharField(max_length=50, null=True)                     # 所属院系
    title = fields.CharField(max_length=50, null=True)                          # 职称（教授/副教授/讲师等）
    # 一对多反向关系：通过 courses 可访问该教师的所有课程
    courses: fields.ReverseRelation["Course"]                                    # 类型标注

    class Meta:
        table = "teachers"                                                      # 指定数据库表名

    def __str__(self):
        return f"Teacher({self.employee_no}, {self.name})"                      # 字符串表示


class Course(Model):
    """课程表 - 存储课程信息（多对一关联教师）"""
    id = fields.BigIntField(pk=True)                                            # 主键，自增 ID
    name = fields.CharField(max_length=100)                                     # 课程名称
    course_no = fields.CharField(max_length=20, unique=True)                    # 课程编号，全局唯一
    teacher = fields.ForeignKeyField("models.Teacher",                          # 多对一关联教师
                                     related_name="courses")                    # 反向关系名（teacher.courses）
    credits = fields.DecimalField(max_digits=3, decimal_places=1)               # 学分（如 3.0）
    max_students = fields.IntField(default=60)                                  # 课程最大容量
    current_count = fields.IntField(default=0)                                  # 当前已选人数
    class_time = fields.CharField(max_length=100, null=True)                    # 上课时间（如"周一 3-4节"）
    classroom = fields.CharField(max_length=50, null=True)                      # 上课教室
    description = fields.TextField(null=True)                                   # 课程简介
    semester = fields.CharField(max_length=20, default="2026春")                # 开课学期
    status = fields.SmallIntField(default=1)                                    # 状态：1=开放选课，0=已关闭

    class Meta:
        table = "courses"                                                       # 指定数据库表名

    def __str__(self):
        return f"Course({self.course_no}, {self.name})"                         # 字符串表示


class Enrollment(Model):
    """选课记录表 - 学生与课程的多对多中间表（附加成绩和状态字段）"""
    id = fields.BigIntField(pk=True)                                            # 主键，自增 ID
    student = fields.ForeignKeyField("models.Student",                          # 多对一关联学生
                                     related_name="enrollments")                # 反向关系名（student.enrollments）
    course = fields.ForeignKeyField("models.Course",                            # 多对一关联课程
                                    related_name="enrollments")                 # 反向关系名（course.enrollments）
    score = fields.DecimalField(max_digits=5, decimal_places=2, null=True)      # 成绩（0-100，两位小数，可空）
    status = fields.SmallIntField(default=1)                                    # 状态：1=正常，0=已退课
    enrolled_at = fields.DatetimeField(auto_now_add=True)                       # 选课时间，自动填充

    class Meta:
        table = "enrollments"                                                   # 指定数据库表名
        unique_together = [("student_id", "course_id")]                         # 联合唯一约束：同一学生不能重复选同一门课

    def __str__(self):
        return f"Enrollment(student={self.student_id}, course={self.course_id})"  # 字符串表示

3.3 关系设计的三个核心问题

问题一：一对一（Student ↔ StudentProfile）

使用 OneToOneField，关键在于创建前检查是否已存在：

@router.post("/{student_id}/profile")
async def create_profile(student_id: int, data: StudentProfileRequest):
    existing = await StudentProfile.filter(student_id=student_id).first()
    if existing:
        raise HTTPException(400, detail="该学生的档案已存在，请使用编辑功能")
    return await StudentProfile.create(student_id=student_id, **data.dict())

问题二：多对一（Course ↔ Teacher）

查询课程列表时，需要用 prefetch_related("teacher") 预加载关联对象，避免 N+1 查询：

courses = await Course.all().prefetch_related("teacher") \
    .offset(offset).limit(page_size)
for c in courses:
    teacher_name = c.teacher.name  # 不触发额外 SQL

问题三：多对多（Student ↔ Course 通过 Enrollment）

选课时需要三重校验：课程状态 → 容量限制 → 重复选课。顺序很重要——先做轻量检查，失败的请求尽早返回：

# 校验 1：课程状态
if course.status != 1:
    raise HTTPException(400, detail="该课程已关闭")

# 校验 2：容量（current_count vs max_students）
if course.current_count >= course.max_students:
    raise HTTPException(400, detail="课程已满")

# 校验 3：唯一约束
existing = await Enrollment.filter(student_id=sid, course_id=cid).first()
if existing:
    raise HTTPException(400, detail="不可重复选课")

---

四、JWT 认证实现

4.1 整体流程

用户登录 → 验证密码 → 签发 JWT → 前端存 localStorage
    ↓
后续请求 → Bearer Token → 依赖注入解析 → 获取当前用户

4.2 Token 签发

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

4.3 依赖注入链

认证依赖设计为链式调用，权限检查复用认证结果：

# 基础认证：解析 Token 并返回用户信息
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    payload = decode_access_token(credentials.credentials)
    if not payload:
        raise HTTPException(401, detail="Token 无效或已过期")
    return payload

# 管理员权限：依赖 get_current_user 的结果再校验 role
async def require_admin(current_user: dict = Depends(get_current_user)):
    if current_user.get("role") != "admin":
        raise HTTPException(403, detail="权限不足")
    return current_user

# 路由中使用
@router.post("/students/")
async def create_student(data: StudentCreate, user=Depends(require_admin)):
    ...  # 仅管理员可执行

这种设计的优势：get_current_user 可复用于所有需要登录的路由，require_admin 仅在管理员接口上叠加调用。

---

五、关键业务逻辑

5.1 选课（事务性操作）

选课涉及两步写入：创建 Enrollment 记录 + 更新 Course.current_count。Tortoise ORM 不原生支持事务装饰器，但核心操作的原子性通过先校验后写入的模式保证：

# 1. 校验阶段（纯读操作，不会产生脏数据）
course = await Course.filter(id=course_id).first()
# ... 三重校验 ...

# 2. 写入阶段（连续写操作）
await Enrollment.create(student_id=sid, course_id=cid)
course.current_count += 1
await course.save()

在生产环境中，建议用 tortoise.transactions 包裹写入操作：

from tortoise.transactions import in_transaction

async with in_transaction():
    await Enrollment.create(...)
    course.current_count += 1
    await course.save()

5.2 级联删除学生

由于 Tortoise ORM 的 ForeignKey 没有 on_delete=CASCADE 的直接支持，需要手动实现级联删除：

async def delete_student(student_id: int):
    # 删除顺序：关联表 → 主表
    await StudentProfile.filter(student_id=student_id).delete()   # 1. 删档案
    await Enrollment.filter(student_id=student_id).delete()       # 2. 删选课记录
    await User.filter(student_id=student_id).delete()             # 3. 删登录账号
    await Student.filter(id=student_id).delete()                  # 4. 删学生

注意：删除顺序必须先子表后主表，否则会因外键约束报错。

5.3 统计分析查询

院系统计——需要跨 teachers 和 courses 两张表聚合：

async def get_department_stats():
    teachers = await Teacher.all().values("id", "department")
    result = {}
    for t in teachers:
        dept = t["department"] or "未分配院系"
        count = await Course.filter(teacher_id=t["id"]).count()
        result[dept] = result.get(dept, 0) + count
    return sorted(result.items(), key=lambda x: -x[1])

成绩分布——纯 Python 分组统计，避免复杂的 CASE WHEN SQL：

async def get_score_distribution():
    enrollments = await Enrollment.filter(
        status=1, score__not_isnull=True
    ).all()
    excellent = sum(1 for e in enrollments if float(e.score) >= 90)
    good = sum(1 for e in enrollments if 80 <= float(e.score) < 90)
    # ...
    return {"excellent": excellent, "good": good, ...}

设计取舍：当数据量较小时（< 10万条），Python 遍历比复杂 SQL 更易维护；数据量大时则应使用 annotate + CASE WHEN 在数据库层面聚合。

---

六、API 接口总览

方法	路径	说明	权限
POST	/api/login	用户登录	公开
GET	/api/students/	学生列表（分页+搜索）	登录
POST	/api/students/	新增学生	管理员
GET	/api/students/{id}	学生详情（含档案+选课）	登录
PUT	/api/students/{id}	编辑学生	管理员
DELETE	/api/students/{id}	删除学生（级联）	管理员
GET	/api/students/{id}/profile	获取档案	登录
POST	/api/students/{id}/profile	创建档案	管理员
PUT	/api/students/{id}/profile	更新档案	管理员
GET	/api/courses/	课程列表（筛选+搜索）	登录
POST	/api/courses/	新增课程	管理员
PUT	/api/courses/{id}	编辑课程	管理员
DELETE	/api/courses/{id}	删除课程	管理员
POST	/api/courses/{id}/enroll	选课	学生
POST	/api/courses/{id}/withdraw	退课	学生
GET	/api/teachers/	教师列表	管理员
POST	/api/teachers/	新增教师	管理员
PUT	/api/teachers/{id}	编辑教师	管理员
DELETE	/api/teachers/{id}	删除教师	管理员
GET	/api/enrollments/	选课记录列表	登录
PUT	/api/enrollments/{id}/score	录入成绩	管理员
GET	/api/stats/overview	概览统计	登录
GET	/api/stats/departments	院系统计	登录
GET	/api/stats/course-ranking	课程排行	登录
GET	/api/stats/score-distribution	成绩分布	登录

---

七、踩坑记录与最佳实践

7.1 Tortoise ORM 的 select_related

查询关联对象时必须显式调用 select_related("teacher")，否则访问 course.teacher.name 会触发额外的异步查询。如果这一行在循环中，就会产生经典的 N+1 查询问题。

7.2 Decimal 字段处理

DecimalField 从数据库读出后是 Decimal 类型，直接扔给 JSON 序列化会报错。需要在响应 Schema 中转换为 float：

class CourseResponse(BaseModel):
    credits: float  # Pydantic 自动将 Decimal → float

    class Config:
        from_attributes = True

7.3 唯一约束的联合索引

Enrollment 表需要 unique_together = [("student_id", "course_id")]，这样数据库层面就能防止重复选课，比纯应用层校验多一重保障。

7.4 密码安全

用户密码使用 bcrypt 算法哈希存储，通过 passlib 库封装：

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(pwd: str) -> str:
    return pwd_context.hash(pwd)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

7.5 异步路由函数

FastAPI 路由函数统一使用 async def，Tortoise ORM 的所有数据库操作都使用 await，确保不阻塞事件循环：

@router.get("/")
async def list_students(page: int = 1):
    students = await Student.all().offset((page-1)*10).limit(10)
    total = await Student.all().count()
    return {"total": total, "list": students}

---

八、总结

本文实现了一个完整的课程选修管理系统后端，核心要点回顾：

1. 一对一关系：创建前检查已存在，避免重复创建档案
2. 多对一关系：select_related() 预加载，防止 N+1 查询
3. 多对多关系：中间表添加 unique_together 约束 + 应用层三重校验
4. 级联删除：手动按外键依赖顺序逐个删除子表记录
5. JWT 认证：链式依赖注入实现「认证 → 授权」分层检查
6. 统计分析：根据数据量权衡 Python 聚合 vs SQL 聚合

完整项目代码可在本地运行，前端基于 Vue3 + Element Plus CDN 构建，可直接用浏览器打开使用。

---

本文由 Claude Code 辅助生成，项目代码结构经过完整验证。