摘要

你是不是写课程大作业时,把所有代码堆在一个 main.py 里,变量名随便起,函数写得像面条,改个登录密码验证到处找?这篇以学生管理系统为例的全流程指南,42000 字详解项目结构规划、README 撰写、高内聚低耦合实现,用 FastAPI 分层架构、依赖注入、接口隔离原则,帮你搭出代码清晰、功能完整、维护简单的项目。


一、引言:为什么要花一周时间规划项目结构?

1.1 课程大作业的常见痛点

回想你之前写的 Python 课程大作业,是不是有以下问题:

  • 代码堆积:所有 API 接口、数据库操作、业务逻辑、工具函数都堆在一个main.pyapp.py里,文件超过 500 行就难以阅读和维护;
  • 变量命名随意:用abctempdata1data2等无意义的变量名,过了一周自己都不知道变量是干什么的;
  • 函数职责不明确:一个函数超过 30 行,既处理 API 请求,又操作数据库,还调用第三方接口,违反单一职责原则;
  • 缺乏测试结构:没有写测试代码,所有功能都靠人工测试,改个小功能要重新测试所有模块;
  • 依赖管理混乱:用pip install安装了一堆库,没有记录依赖版本,换个环境部署时出现版本不兼容的问题;
  • README 简陋:只有一句 “这是一个学生管理系统”,没有部署指南、使用说明、核心功能介绍。
1.2 规划项目结构的好处

花一周时间规划项目结构,可以解决以上所有问题:

  • 代码清晰易读:分层架构让代码职责明确,每个文件的功能单一,变量命名规范,函数长度适中;
  • 维护简单高效:修改一个功能时,只需要修改对应的模块,不会影响其他模块;
  • 测试覆盖全面:分层架构让测试变得简单,每个模块可以独立测试,提高测试覆盖率;
  • 部署方便快捷:依赖管理规范,换个环境部署时只需要安装对应的依赖库;
  • 协作开发顺畅:分层架构让团队协作变得简单,每个成员负责一个模块,不会相互干扰。

二、基础概念扫盲

2.1 项目架构

项目架构是指项目的整体结构和组件之间的关系,它决定了项目的可扩展性、可维护性、可测试性。常见的项目架构有:

  • MVC 架构:Model(数据模型)、View(视图)、Controller(控制器)三层架构,是 Web 开发中最常用的架构之一;
  • MVT 架构:Model(数据模型)、View(视图函数)、Template(模板)三层架构,是 Django 框架的架构;
  • Clean Architecture:由内到外分为实体层、用例层、接口适配器层、框架和驱动层,是一种更清洁、更易维护的架构;
  • DDD 架构:领域驱动设计架构,是一种以业务领域为核心的架构,适合复杂的业务系统。
2.2 高内聚

高内聚是指一个模块内部的元素(变量、函数、类)之间的关联性很强,它们共同完成一个单一的功能。比如学生管理系统中的用户管理模块,内部的元素(用户注册函数、用户登录函数、用户查询函数、用户更新函数、用户删除函数)之间的关联性很强,共同完成用户管理的功能。

2.3 低耦合

低耦合是指模块之间的关联性很弱,一个模块的修改不会影响其他模块。比如学生管理系统中的用户管理模块和课程管理模块之间的关联性很弱,修改用户管理模块的代码不会影响课程管理模块的功能。

2.4 解耦

解耦是指降低模块之间的关联性,让每个模块尽可能独立。常见的解耦方法有:

  • 依赖注入:将模块的依赖对象通过参数传递的方式注入到模块内部,而不是在模块内部直接创建依赖对象;
  • 接口隔离原则:将一个大的接口分成多个小的接口,每个接口只包含一个单一的功能;
  • 策略模式:定义一系列算法,将每个算法封装成一个类,让它们可以相互替换;
  • 观察者模式:定义对象之间的一对多依赖关系,当一个对象状态改变时,所有依赖它的对象都会自动更新。

三、项目结构规划流程

3.1 需求分析

需求分析是项目结构规划的第一步,它决定了项目的功能模块和技术选型。需求分析可以分为功能需求和非功能需求。

3.1.1 功能需求

学生管理系统的功能需求可以分为以下几个模块:

  • 用户管理模块:用户注册、用户登录、用户查询、用户更新、用户删除;
  • 学生管理模块:学生信息查询、学生信息添加、学生信息更新、学生信息删除;
  • 教师管理模块:教师信息查询、教师信息添加、教师信息更新、教师信息删除;
  • 课程管理模块:课程信息查询、课程信息添加、课程信息更新、课程信息删除;
  • 班级管理模块:班级信息查询、班级信息添加、班级信息更新、班级信息删除;
  • 选课管理模块:学生选课、学生退课、教师查询选课情况、管理员查询选课情况;
  • 成绩管理模块:教师录入成绩、学生查询成绩、教师查询成绩、管理员查询成绩;
  • 权限管理模块:用户角色管理、用户权限分配、用户角色查询、用户权限查询。
3.1.2 非功能需求

学生管理系统的非功能需求可以分为以下几个方面:

  • 性能需求:单个 API 接口的响应时间不超过 2 秒,并发用户数不低于 100;
  • 安全需求:用户密码加密存储,API 接口支持 JWT 身份验证,API 接口限流;
  • 可维护性需求:代码清晰易读,分层架构,每个模块职责明确;
  • 可测试性需求:每个模块可以独立测试,测试覆盖率不低于 80%;
  • 可扩展性需求:项目架构支持新增功能模块,比如新增考勤管理模块、新增作业管理模块;
  • 部署需求:项目支持本地部署和 Docker 部署。
3.2 功能模块划分

根据需求分析的结果,将学生管理系统划分为以下几个功能模块:

  • 用户管理模块:负责用户的注册、登录、查询、更新、删除;
  • 学生管理模块:负责学生信息的查询、添加、更新、删除;
  • 教师管理模块:负责教师信息的查询、添加、更新、删除;
  • 课程管理模块:负责课程信息的查询、添加、更新、删除;
  • 班级管理模块:负责班级信息的查询、添加、更新、删除;
  • 选课管理模块:负责学生选课、学生退课、教师查询选课情况、管理员查询选课情况;
  • 成绩管理模块:负责教师录入成绩、学生查询成绩、教师查询成绩、管理员查询成绩;
  • 权限管理模块:负责用户角色管理、用户权限分配、用户角色查询、用户权限查询。
3.3 技术选型

根据需求分析的结果,选择以下技术栈:

技术栈 用途 版本
FastAPI Web 框架 0.109.0
Uvicorn ASGI 服务器 0.27.0
Gunicorn WSGI 服务器 21.2.0
MySQL 关系型数据库 8.0
SQLAlchemy ORM 框架 2.0.25
Alembic 数据库迁移工具 1.13.1
Python-multipart 处理表单数据 0.0.6
Python-jose[cryptography] 生成和验证 JWT 令牌 3.3.0
Passlib[bcrypt] 加密和验证用户密码 1.7.4
Pytest 单元测试框架 7.4.4
Pytest-cov 测试覆盖率工具 4.1.0
Docker 容器化部署工具 20.10.23
Docker Compose 容器编排工具 2.15.1
Nginx 反向代理服务器 1.24.0
3.4 目录树设计

根据功能模块划分和技术选型的结果,设计以下项目目录树:

student-management-system/
├── .github/
│   ├── workflows/
│   │   └── ci.yml  # GitHub Actions CI/CD配置
├── app/
│   ├── api/
│   │   ├── __init__.py
│   │   ├── endpoints/
│   │   │   ├── __init__.py
│   │   │   ├── auth.py  # 身份验证接口
│   │   │   ├── users.py  # 用户管理接口
│   │   │   ├── students.py  # 学生管理接口
│   │   │   ├── teachers.py  # 教师管理接口
│   │   │   ├── courses.py  # 课程管理接口
│   │   │   ├── classes.py  # 班级管理接口
│   │   │   ├── enrollments.py  # 选课管理接口
│   │   │   ├── grades.py  # 成绩管理接口
│   │   │   └── roles.py  # 权限管理接口
│   │   └── schemas/
│   │       ├── __init__.py
│   │       ├── auth.py  # 身份验证接口的数据验证模型
│   │       ├── users.py  # 用户管理接口的数据验证模型
│   │       ├── students.py  # 学生管理接口的数据验证模型
│   │       ├── teachers.py  # 教师管理接口的数据验证模型
│   │       ├── courses.py  # 课程管理接口的数据验证模型
│   │       ├── classes.py  # 班级管理接口的数据验证模型
│   │       ├── enrollments.py  # 选课管理接口的数据验证模型
│   │       ├── grades.py  # 成绩管理接口的数据验证模型
│   │       └── roles.py  # 权限管理接口的数据验证模型
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py  # 项目配置
│   │   ├── database.py  # 数据库连接配置
│   │   ├── security.py  # 安全配置(密码加密/验证、JWT生成/验证)
│   │   └── dependencies.py  # 依赖注入函数
│   ├── models/
│   │   ├── __init__.py
│   │   ├── users.py  # 用户信息的ORM模型
│   │   ├── students.py  # 学生信息的ORM模型
│   │   ├── teachers.py  # 教师信息的ORM模型
│   │   ├── courses.py  # 课程信息的ORM模型
│   │   ├── classes.py  # 班级信息的ORM模型
│   │   ├── enrollments.py  # 选课信息的ORM模型
│   │   ├── grades.py  # 成绩信息的ORM模型
│   │   └── roles.py  # 权限信息的ORM模型
│   ├── services/
│   │   ├── __init__.py
│   │   ├── auth_service.py  # 身份验证业务逻辑
│   │   ├── users_service.py  # 用户管理业务逻辑
│   │   ├── students_service.py  # 学生管理业务逻辑
│   │   ├── teachers_service.py  # 教师管理业务逻辑
│   │   ├── courses_service.py  # 课程管理业务逻辑
│   │   ├── classes_service.py  # 班级管理业务逻辑
│   │   ├── enrollments_service.py  # 选课管理业务逻辑
│   │   ├── grades_service.py  # 成绩管理业务逻辑
│   │   └── roles_service.py  # 权限管理业务逻辑
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── logger.py  # 日志管理工具
│   │   ├── exceptions.py  # 自定义异常类
│   │   └── helpers.py  # 辅助工具函数
│   └── main.py  # FastAPI应用的主入口文件
├── alembic/  # Alembic数据库迁移工具的配置和迁移脚本
├── tests/
│   ├── __init__.py
│   ├── conftest.py  # 测试配置和共享函数
│   ├── test_auth.py  # 身份验证接口的测试
│   ├── test_users.py  # 用户管理接口的测试
│   ├── test_students.py  # 学生管理接口的测试
│   ├── test_teachers.py  # 教师管理接口的测试
│   ├── test_courses.py  # 课程管理接口的测试
│   ├── test_classes.py  # 班级管理接口的测试
│   ├── test_enrollments.py  # 选课管理接口的测试
│   ├── test_grades.py  # 成绩管理接口的测试
│   └── test_roles.py  # 权限管理接口的测试
├── static/  # 静态资源(CSS/JS/images,可选)
├── templates/  # 模板文件(HTML,可选)
├── .dockerignore  # Docker忽略文件
├── .gitignore  # Git忽略文件
├── alembic.ini  # Alembic配置文件
├── docker-compose.yml  # Docker Compose配置文件
├── Dockerfile.fastapi  # FastAPI的Dockerfile配置文件
├── Dockerfile.nginx  # Nginx的Dockerfile配置文件
├── requirements.dev.txt  # 开发环境所需的依赖库
├── requirements.prod.txt  # 生产环境所需的依赖库
└── README.md  # 项目说明文档
3.5 文件命名规范

为了让项目的代码清晰易读,需要遵循以下文件命名规范:

  • Python 文件:使用小写字母和下划线组成,如auth.pyusers_service.py
  • Python 类:使用大驼峰命名法,如UserCreateUserInfo
  • Python 函数:使用小写字母和下划线组成,如get_current_active_usercreate_user
  • Python 变量:使用小写字母和下划线组成,如user_idusername
  • 数据库表:使用小写字母和下划线组成,如usersstudents
  • 数据库字段:使用小写字母和下划线组成,如user_idusername
  • 配置文件:使用小写字母和下划线组成,如config.pyalembic.ini
  • 测试文件:使用test_前缀和对应的模块名组成,如test_auth.pytest_users.py

四、分层架构详解

根据项目目录树设计的结果,学生管理系统采用Clean Architecture的简化版本,分为以下几个层次:

4.1 实体层(Models)

实体层是项目的核心层,包含了项目的业务实体和数据模型。实体层的代码不依赖于任何外部框架,只包含业务逻辑的核心部分。比如学生管理系统中的用户实体、学生实体、教师实体、课程实体、班级实体、选课实体、成绩实体、权限实体。

打开app/models/users.py文件,输入以下代码:

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from app.core.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(20), unique=True, index=True, nullable=False)
    email = Column(String(50), unique=True, index=True, nullable=False)
    hashed_password = Column(String(100), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

打开app/models/students.py文件,输入以下代码:

from sqlalchemy import Column, Integer, String, Date, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.core.database import Base

class Student(Base):
    __tablename__ = "students"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), unique=True, nullable=False)
    student_no = Column(String(20), unique=True, index=True, nullable=False)
    name = Column(String(50), nullable=False)
    gender = Column(String(10), nullable=False)
    birthdate = Column(Date, nullable=True)
    phone = Column(String(20), nullable=True)
    email = Column(String(50), nullable=True)
    address = Column(String(200), nullable=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    user = relationship("User", back_populates="student")
    enrollments = relationship("Enrollment", back_populates="student")
    grades = relationship("Grade", back_populates="student")
4.2 接口适配器层(API)

接口适配器层负责将实体层的业务实体转换为 API 接口的数据验证模型,处理 API 请求和响应,调用服务层的业务逻辑。接口适配器层的代码依赖于 FastAPI 框架和 Pydantic 库。比如学生管理系统中的用户管理接口、学生管理接口、教师管理接口、课程管理接口、班级管理接口、选课管理接口、成绩管理接口、权限管理接口。

打开app/api/schemas/users.py文件,输入以下代码:

from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$")
    email: EmailStr

class UserCreate(UserBase):
    password: str = Field(..., min_length=8, max_length=20, 
                          regex=r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).*$")

class UserUpdate(BaseModel):
    username: Optional[str] = Field(None, min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$")
    email: Optional[EmailStr] = None
    password: Optional[str] = Field(None, min_length=8, max_length=20, 
                                   regex=r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).*$")
    is_active: Optional[bool] = None

class UserInfo(UserBase):
    id: int
    is_active: bool = True
    created_at: datetime
    updated_at: Optional[datetime] = None

    class Config:
        orm_mode = True

打开app/api/endpoints/users.py文件,输入以下代码:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.dependencies import get_current_active_user, get_current_active_admin_user
from app.models.users import User
from app.schemas.users import UserCreate, UserUpdate, UserInfo
from app.services.users_service import (
    get_user_by_id,
    get_user_by_username,
    get_user_by_email,
    get_all_users,
    create_user,
    update_user,
    delete_user
)
from typing import List, Optional

router = APIRouter(prefix="/users", tags=["Users"])

@router.get("/", response_model=List[UserInfo])
async def read_users(page: int = 1, page_size: int = 10, is_active: Optional[bool] = None,
                     db: Session = Depends(get_db), current_user: User = Depends(get_current_active_admin_user)):
    try:
        users = get_all_users(db, page, page_size, is_active)
        return users
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取用户列表失败:{e}")

@router.get("/{user_id}", response_model=UserInfo)
async def read_user(user_id: int, db: Session = Depends(get_db),
                    current_user: User = Depends(get_current_active_user)):
    try:
        user = get_user_by_id(db, user_id)
        if not user:
            raise HTTPException(status_code=404, detail="用户不存在")
        
        if not current_user.is_admin and current_user.id != user_id:
            raise HTTPException(status_code=403, detail="没有权限访问该用户信息")
        
        return user
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取用户信息失败:{e}")

@router.post("/", response_model=UserInfo)
async def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db),
                               current_user: User = Depends(get_current_active_admin_user)):
    try:
        existing_user = get_user_by_username(db, user.username)
        if existing_user:
            raise HTTPException(status_code=400, detail="用户名已存在")
        
        existing_email = get_user_by_email(db, user.email)
        if existing_email:
            raise HTTPException(status_code=400, detail="邮箱已存在")
        
        new_user = create_user(db, user)
        return new_user
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"创建用户失败:{e}")

@router.put("/{user_id}", response_model=UserInfo)
async def update_user_endpoint(user_id: int, user_update: UserUpdate, db: Session = Depends(get_db),
                               current_user: User = Depends(get_current_active_user)):
    try:
        user = get_user_by_id(db, user_id)
        if not user:
            raise HTTPException(status_code=404, detail="用户不存在")
        
        if not current_user.is_admin and current_user.id != user_id:
            raise HTTPException(status_code=403, detail="没有权限修改该用户信息")
        
        if user_update.username and user_update.username != user.username:
            existing_user = get_user_by_username(db, user_update.username)
            if existing_user:
                raise HTTPException(status_code=400, detail="用户名已存在")
        
        if user_update.email and user_update.email != user.email:
            existing_email = get_user_by_email(db, user_update.email)
            if existing_email:
                raise HTTPException(status_code=400, detail="邮箱已存在")
        
        updated_user = update_user(db, user_id, user_update)
        return updated_user
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"修改用户信息失败:{e}")

@router.delete("/{user_id}")
async def delete_user_endpoint(user_id: int, db: Session = Depends(get_db),
                               current_user: User = Depends(get_current_active_admin_user)):
    try:
        user = get_user_by_id(db, user_id)
        if not user:
            raise HTTPException(status_code=404, detail="用户不存在")
        
        delete_user(db, user_id)
        return {"message": "用户删除成功"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"删除用户失败:{e}")
4.3 服务层(Services)

服务层负责处理项目的业务逻辑,调用实体层的业务实体,与数据库进行交互。服务层的代码依赖于 SQLAlchemy ORM 框架,不依赖于任何外部框架。比如学生管理系统中的用户管理业务逻辑、学生管理业务逻辑、教师管理业务逻辑、课程管理业务逻辑、班级管理业务逻辑、选课管理业务逻辑、成绩管理业务逻辑、权限管理业务逻辑。

打开app/services/users_service.py文件,输入以下代码:

from sqlalchemy.orm import Session
from app.models.users import User
from app.schemas.users import UserCreate, UserUpdate
from app.core.security import get_password_hash, verify_password

def get_user_by_id(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

def get_user_by_username(db: Session, username: str):
    return db.query(User).filter(User.username == username).first()

def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

def get_all_users(db: Session, page: int = 1, page_size: int = 10, is_active: bool = None):
    query = db.query(User)
    if is_active is not None:
        query = query.filter(User.is_active == is_active)
    
    start_index = (page - 1) * page_size
    end_index = start_index + page_size
    return query.offset(start_index).limit(page_size).all()

def create_user(db: Session, user: UserCreate):
    hashed_password = get_password_hash(user.password)
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user_id: int, user_update: UserUpdate):
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        return None
    
    if user_update.username:
        db_user.username = user_update.username
    if user_update.email:
        db_user.email = user_update.email
    if user_update.password:
        db_user.hashed_password = get_password_hash(user_update.password)
    if user_update.is_active is not None:
        db_user.is_active = user_update.is_active
    
    db.commit()
    db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    db_user = get_user_by_id(db, user_id)
    if db_user:
        db.delete(db_user)
        db.commit()
4.4 框架和驱动层(Core)

框架和驱动层负责处理项目的框架和驱动相关的配置,如数据库连接配置、项目配置、安全配置、依赖注入函数。框架和驱动层的代码依赖于 FastAPI 框架、SQLAlchemy ORM 框架、Python-jose [cryptography] 库、Passlib [bcrypt] 库。

打开app/core/config.py文件,输入以下代码:

from pydantic import BaseSettings

class Settings(BaseSettings):
    # 项目配置
    PROJECT_NAME: str = "学生管理系统"
    PROJECT_VERSION: str = "1.0.0"
    PROJECT_DESCRIPTION: str = "一个基于FastAPI的学生管理系统"
    
    # 数据库配置
    DATABASE_URL: str = "mysql+pymysql://root:123456@localhost/student_management_system"
    
    # JWT配置
    SECRET_KEY: str = "your-secret-key-here"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    # 限流配置
    RATE_LIMIT: str = "100/minute"
    
    class Config:
        env_file = ".env"

settings = Settings()

打开app/core/database.py文件,输入以下代码:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

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

五、README 撰写详细步骤 + 示例

README 是项目的说明文档,是项目的 “门面”,它决定了用户对项目的第一印象。README 的内容要详细、完整、清晰、易读,包括项目简介、技术栈、快速开始、核心功能、使用指南、开发指南、贡献指南、许可证、联系方式、FAQ 等部分。

5.1 README 的结构

README 的结构可以分为以下几个部分:

  1. 项目简介:项目的功能、特点、适用场景;
  2. 技术栈:项目所需的技术栈;
  3. 快速开始:项目的本地部署和 Docker 部署指南;
  4. 核心功能:项目的核心功能介绍;
  5. 使用指南:项目的使用方法;
  6. 开发指南:项目的开发环境搭建和开发流程;
  7. 贡献指南:项目的贡献方法;
  8. 许可证:项目的许可证;
  9. 联系方式:项目的联系方式;
  10. FAQ:项目的常见问题与解决方案。
5.2 README 的撰写注意事项
  1. 内容详细完整:README 的内容要详细、完整,包括项目的所有核心功能、使用方法、开发流程、部署指南;
  2. 语言通俗易懂:README 的语言要通俗易懂,面向初学者,避免使用太专业的术语;
  3. 格式清晰易读:README 的格式要清晰易读,使用 Markdown 语法,添加标题、列表、表格、代码块、图片等;
  4. 图片和超链接:README 的内容要添加图片和超链接,让用户更容易理解项目的功能和使用方法;
  5. 代码示例:README 的内容要添加代码示例,让用户更容易理解项目的开发流程和部署指南;
  6. 更新及时:README 的内容要随着项目的发展及时更新,避免过时的内容。
5.3 README 的示例
# 学生管理系统

## 项目简介
这是一个基于FastAPI的学生管理系统,它提供了用户管理、学生管理、教师管理、课程管理、班级管理、选课管理、成绩管理、权限管理等功能。系统采用分层架构,代码清晰易读,功能完整,维护简单。

## 技术栈
| 技术栈 | 用途 | 版本 |
| --- | --- | --- |
| FastAPI | Web框架 | 0.109.0 |
| Uvicorn | ASGI服务器 | 0.27.0 |
| Gunicorn | WSGI服务器 | 21.2.0 |
| MySQL | 关系型数据库 | 8.0 |
| SQLAlchemy | ORM框架 | 2.0.25 |
| Alembic | 数据库迁移工具 | 1.13.1 |
| Python-multipart | 处理表单数据 | 0.0.6 |
| Python-jose[cryptography] | 生成和验证JWT令牌 | 3.3.0 |
| Passlib[bcrypt] | 加密和验证用户密码 | 1.7.4 |
| Pytest | 单元测试框架 | 7.4.4 |
| Pytest-cov | 测试覆盖率工具 | 4.1.0 |
| Docker | 容器化部署工具 | 20.10.23 |
| Docker Compose | 容器编排工具 | 2.15.1 |
| Nginx | 反向代理服务器 | 1.24.0 |

## 快速开始

### 本地部署

#### 1. 安装依赖库
```bash
pip install -r requirements.dev.txt
2. 配置数据库
  • 安装 MySQL Server:https://dev.mysql.com/downloads/mysql/
  • 启动 MySQL Server
  • 创建数据库:
    CREATE DATABASE student_management_system CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
    
3. 配置环境变量

创建一个名为.env的文件,内容如下:

DATABASE_URL=mysql+pymysql://root:123456@localhost/student_management_system
SECRET_KEY=your-secret-key-here
4. 数据库迁移
alembic upgrade head
5. 启动开发服务器
uvicorn app.main:app --reload
6. 测试项目

Docker 部署

1. 安装 Docker 和 Docker Compose
2. 配置环境变量

创建一个名为.env的文件,内容如下:

DATABASE_URL=mysql+pymysql://root:123456@mysql/student_management_system
SECRET_KEY=your-secret-key-here
MYSQL_ROOT_PASSWORD=123456
MYSQL_DATABASE=student_management_system
3. 启动项目
docker-compose up -d
4. 数据库迁移
docker exec -it fastapi-app alembic upgrade head
5. 测试项目

核心功能

1. 用户管理功能

  • 用户注册:新用户可以通过用户名、邮箱、密码注册;
  • 用户登录:已注册用户可以通过用户名和密码登录,获取 JWT 访问令牌;
  • 用户查询:管理员可以查询所有用户的信息,普通用户可以查询自己的信息;
  • 用户更新:管理员可以更新所有用户的信息,普通用户可以更新自己的信息;
  • 用户删除:管理员可以删除所有用户的信息,普通用户可以删除自己的信息。

2. 学生管理功能

  • 学生信息查询:管理员和教师可以查询所有学生的信息,学生可以查询自己的信息;
  • 学生信息添加:管理员可以添加学生的信息;
  • 学生信息更新:管理员和教师可以更新学生的信息,学生可以更新自己的信息;
  • 学生信息删除:管理员可以删除学生的信息。

3. 教师管理功能

  • 教师信息查询:管理员可以查询所有教师的信息,教师可以查询自己的信息;
  • 教师信息添加:管理员可以添加教师的信息;
  • 教师信息更新:管理员和教师可以更新教师的信息,教师可以更新自己的信息;
  • 教师信息删除:管理员可以删除教师的信息。

4. 课程管理功能

  • 课程信息查询:管理员、教师、学生可以查询所有课程的信息;
  • 课程信息添加:管理员和教师可以添加课程的信息;
  • 课程信息更新:管理员和教师可以更新课程的信息;
  • 课程信息删除:管理员和教师可以删除课程的信息。

5. 班级管理功能

  • 班级信息查询:管理员、教师、学生可以查询所有班级的信息;
  • 班级信息添加:管理员可以添加班级的信息;
  • 班级信息更新:管理员和教师可以更新班级的信息;
  • 班级信息删除:管理员可以删除班级的信息。

6. 选课管理功能

  • 学生选课:学生可以选择自己感兴趣的课程;
  • 学生退课:学生可以退选自己已经选择的课程;
  • 教师查询选课情况:教师可以查询自己所教课程的选课情况;
  • 管理员查询选课情况:管理员可以查询所有课程的选课情况。

7. 成绩管理功能

  • 教师录入成绩:教师可以录入自己所教课程的学生成绩;
  • 学生查询成绩:学生可以查询自己的成绩;
  • 教师查询成绩:教师可以查询自己所教课程的学生成绩;
  • 管理员查询成绩:管理员可以查询所有课程的学生成绩。

8. 权限管理功能

  • 用户角色管理:管理员可以添加、查询、更新、删除用户角色;
  • 用户权限分配:管理员可以为用户分配角色;
  • 用户角色查询:管理员可以查询用户的角色;
  • 用户权限查询:管理员可以查询用户的权限。

使用指南

1. 用户注册

打开浏览器,访问 http://127.0.0.1:8000/docs,点击/auth/register接口旁边的Try it out按钮,输入以下 JSON 数据:

{
  "username": "testuser",
  "email": "testuser@example.com",
  "password": "Test1234"
}

点击Execute按钮,返回以下 JSON 响应:

{
  "id": 1,
  "username": "testuser",
  "email": "testuser@example.com",
  "is_active": true,
  "created_at": "2025-01-01T10:00:00",
  "updated_at": null
}

2. 用户登录

打开浏览器,访问 http://127.0.0.1:8000/docs,点击/auth/login接口旁边的Try it out按钮,输入以下 JSON 数据:

{
  "username": "testuser",
  "password": "Test1234"
}

点击Execute按钮,返回以下 JSON 响应:

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ0ZXN0dXNlciIsImV4cCI6MTczMTczNjUxMX0.abc123",
  "token_type": "bearer"
}

3. 查询用户信息

打开浏览器,访问 http://127.0.0.1:8000/docs,点击右上角的Authorize按钮,输入Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ0ZXN0dXNlciIsImV4cCI6MTczMTczNjUxMX0.abc123,点击Authorize按钮,点击Close按钮。点击/users/me接口旁边的Try it out按钮,点击Execute按钮,返回以下 JSON 响应:

{
  "id": 1,
  "username": "testuser",
  "email": "testuser@example.com",
  "is_active": true,
  "created_at": "2025-01-01T10:00:00",
  "updated_at": null
}

开发指南

1. 开发环境搭建

2. 开发流程

  • 克隆项目:git clone https://github.com/your-username/student-management-system.git
  • 进入项目目录:cd student-management-system
  • 创建虚拟环境:python -m venv venv
  • 激活虚拟环境:
    • Windows:venv\Scripts\activate
    • Mac/Linux:source venv/bin/activate
  • 安装依赖库:pip install -r requirements.dev.txt
  • 配置数据库:创建一个名为.env的文件,内容如下:
    DATABASE_URL=mysql+pymysql://root:123456@localhost/student_management_system
    SECRET_KEY=your-secret-key-here
    
  • 数据库迁移:alembic upgrade head
  • 启动开发服务器:uvicorn app.main:app --reload
  • 编写代码:在app/目录下编写代码
  • 测试代码:在tests/目录下编写测试代码,运行pytest命令测试代码
  • 提交代码:使用 Git 提交代码到 GitHub

3. 项目结构

项目的目录树如下:

student-management-system/
├── .github/
│   ├── workflows/
│   │   └── ci.yml  # GitHub Actions CI/CD配置
├── app/
│   ├── api/
│   │   ├── __init__.py
│   │   ├── endpoints/
│   │   │   ├── __init__.py
│   │   │   ├── auth.py  # 身份验证接口
│   │   │   ├── users.py  # 用户管理接口
│   │   │   ├── students.py  # 学生管理接口
│   │   │   ├── teachers.py  # 教师管理接口
│   │   │   ├── courses.py  # 课程管理接口
│   │   │   ├── classes.py  # 班级管理接口
│   │   │   ├── enrollments.py  # 选课管理接口
│   │   │   ├── grades.py  # 成绩管理接口
│   │   │   └── roles.py  # 权限管理接口
│   │   └── schemas/
│   │       ├── __init__.py
│   │       ├── auth.py  # 身份验证接口的数据验证模型
│   │       ├── users.py  # 用户管理接口的数据验证模型
│   │       ├── students.py  # 学生管理接口的数据验证模型
│   │       ├── teachers.py  # 教师管理接口的数据验证模型
│   │       ├── courses.py  # 课程管理接口的数据验证模型
│   │       ├── classes.py  # 班级管理接口的数据验证模型
│   │       ├── enrollments.py  # 选课管理接口的数据验证模型
│   │       ├── grades.py  # 成绩管理接口的数据验证模型
│   │       └── roles.py  # 权限管理接口的数据验证模型
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py  # 项目配置
│   │   ├── database.py  # 数据库连接配置
│   │   ├── security.py  # 安全配置(密码加密/验证、JWT生成/验证)
│   │   └── dependencies.py  # 依赖注入函数
│   ├── models/
│   │   ├── __init__.py
│   │   ├── users.py  # 用户信息的ORM模型
│   │   ├── students.py  # 学生信息的ORM模型
│   │   ├── teachers.py  # 教师信息的ORM模型
│   │   ├── courses.py  # 课程信息的ORM模型
│   │   ├── classes.py  # 班级管理接口的数据验证模型
│   │   ├── enrollments.py  # 选课管理接口的数据验证模型
│   │   ├── grades.py  # 成绩管理接口的数据验证模型
│   │   └── roles.py  # 权限管理接口的数据验证模型
│   ├── services/
│   │   ├── __init__.py
│   │   ├── auth_service.py  # 身份验证业务逻辑
│   │   ├── users_service.py  # 用户管理业务逻辑
│   │   ├── students_service.py  # 学生管理业务逻辑
│   │   ├── teachers_service.py  # 教师管理业务逻辑
│   │   ├── courses_service.py  # 课程管理业务逻辑
│   │   ├── classes_service.py  # 班级管理业务逻辑
│   │   ├── enrollments_service.py  # 选课管理业务逻辑
│   │   ├── grades_service.py  # 成绩管理业务逻辑
│   │   └── roles_service.py  # 权限管理业务逻辑
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── logger.py  # 日志管理工具
│   │   ├── exceptions.py  # 自定义异常类
│   │   └── helpers.py  # 辅助工具函数
│   └── main.py  # FastAPI应用的主入口文件
├── alembic/  # Alembic数据库迁移工具的配置和迁移脚本
├── tests/
│   ├── __init__.py
│   ├── conftest.py  # 测试配置和共享函数
│   ├── test_auth.py  # 身份验证接口的测试
│   ├── test_users.py  # 用户管理接口的测试
│   ├── test_students.py  # 学生管理接口的测试
│   ├── test_teachers.py  # 教师管理接口的测试
│   ├── test_courses.py  # 课程管理接口的测试
│   ├── test_classes.py  # 班级管理接口的测试
│   ├── test_enrollments.py  # 选课管理接口的测试
│   ├── test_grades.py  # 成绩管理接口的测试
│   └── test_roles.py  # 权限管理接口的测试
├── static/  # 静态资源(CSS/JS/images,可选)
├── templates/  # 模板文件(HTML,可选)
├── .dockerignore  # Docker忽略文件
├── .gitignore  # Git忽略文件
├── alembic.ini  # Alembic配置文件
├── docker-compose.yml  # Docker Compose配置文件
├── Dockerfile.fastapi  # FastAPI的Dockerfile配置文件
├── Dockerfile.nginx  # Nginx的Dockerfile配置文件
├── requirements.dev.txt  # 开发环境所需的依赖库
├── requirements.prod.txt  # 生产环境所需的依赖库
└── README.md  # 项目说明文档

贡献指南

1. 提交代码规范

  • 提交信息要清晰、简洁,说明提交的内容;
  • 提交信息的格式:<type>(<scope>): <subject>
  • 提交类型(type):
    • feat:新增功能;
    • fix:修复 bug;
    • docs:文档更新;
    • style:代码格式调整;
    • refactor:代码重构;
    • test:测试代码更新;
    • chore:其他更新;
  • 提交范围(scope):提交的模块或功能;
  • 提交主题(subject):提交的内容;

2. 代码规范

3. 测试规范

  • 每个模块都要写测试代码;
  • 测试代码要覆盖所有的业务逻辑;
  • 测试代码要使用 pytest 框架;
  • 测试代码的文件名要使用test_前缀和对应的模块名组成;
  • 测试函数的文件名要使用test_前缀和对应的函数名组成。

许可证

MIT License

联系方式

FAQ

1. 如何修改数据库连接配置?

修改.env文件中的DATABASE_URL变量,格式如下:

DATABASE_URL=mysql+pymysql://<username>:<password>@<host>/<database>

2. 如何修改 JWT 配置?

修改.env文件中的SECRET_KEYACCESS_TOKEN_EXPIRE_MINUTES变量。

3. 如何添加新的功能模块?

  • app/models/目录下创建新的 ORM 模型文件;
  • app/schemas/目录下创建新的 Pydantic 数据验证模型文件;
  • app/services/目录下创建新的业务逻辑文件;
  • app/api/endpoints/目录下创建新的 API 接口文件;
  • tests/目录下创建新的测试代码文件。

4. 如何部署到生产环境?

使用 Docker Compose 部署到生产环境,修改.env文件中的配置,运行docker-compose up -d命令。


---

### 六、依赖管理
依赖管理是项目的重要组成部分,它决定了项目的可部署性和可维护性。依赖管理的目标是确保项目在不同的环境中使用相同版本的依赖库,避免版本不兼容的问题。

#### 6.1 依赖管理工具对比
常见的Python依赖管理工具有`venv`、`poetry`、`pipenv`,它们的优缺点如下:

| 依赖管理工具 | 优点 | 缺点 | 适用场景 |
| --- | --- | --- | --- |
| venv | Python内置,无需安装额外依赖库,操作简单 | 不支持依赖锁定,不支持开发依赖和生产依赖分离 | 简单的Python项目 |
| poetry | 支持依赖锁定,支持开发依赖和生产依赖分离,操作简单 | 学习曲线较陡,安装时间长 | 复杂的Python项目 |
| pipenv | 支持依赖锁定,支持开发依赖和生产依赖分离,操作简单 | 学习曲线较陡,安装时间长 | 复杂的Python项目 |

#### 6.2 依赖管理规范
为了确保项目的依赖管理规范,需要遵循以下几点:
- 使用`requirements.dev.txt`和`requirements.prod.txt`分离开发依赖和生产依赖;
- 每个依赖库的版本要明确;
- 定期更新依赖库的版本;
- 使用依赖锁定工具(如`pip freeze`、`poetry lock`、`pipenv lock`)锁定依赖库的版本。

#### 6.3 依赖管理代码示例
打开`requirements.dev.txt`文件,输入以下代码:

fastapi==0.109.0uvicorn==0.27.0python-multipart==0.0.6python-jose[cryptography]==3.3.0passlib[bcrypt]==1.7.4sqlalchemy==2.0.25alembic==1.13.1pymysql==1.1.0pytest==7.4.4pytest-cov==4.1.0


打开`requirements.prod.txt`文件,输入以下代码:

fastapi==0.109.0uvicorn==0.27.0gunicorn==21.2.0python-multipart==0.0.6python-jose[cryptography]==3.3.0passlib[bcrypt]==1.7.4sqlalchemy==2.0.25alembic==1.13.1pymysql==1.1.0


---

### 七、日志管理
日志管理是项目的重要组成部分,它决定了项目的可维护性和可调试性。日志管理的目标是记录项目的运行状态和错误信息,方便开发者排查问题和调试代码。

#### 7.1 日志级别
常见的Python日志级别有以下几种:
- `DEBUG`:调试信息,用于排查问题和调试代码;
- `INFO`:普通信息,用于记录项目的运行状态;
- `WARNING`:警告信息,用于记录项目的警告事件;
- `ERROR`:错误信息,用于记录项目的错误事件;
- `CRITICAL`:严重错误信息,用于记录项目的严重错误事件。

#### 7.2 日志配置
打开`app/utils/logger.py`文件,输入以下代码:
```python
import logging
import sys
from logging.handlers import RotatingFileHandler

def setup_logger(name: str = "student-management-system", level: int = logging.INFO) -> logging.Logger:
    """
    配置日志管理工具
    :param name: 日志管理器的名称
    :param level: 日志级别
    :return: 日志管理器
    """
    logger = logging.getLogger(name)
    logger.setLevel(level)
    logger.propagate = False

    # 控制台输出格式化器
    console_formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )

    # 文件输出格式化器
    file_formatter = logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s - %(pathname)s - %(lineno)d"
    )

    # 控制台输出处理器
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(level)
    console_handler.setFormatter(console_formatter)
    logger.addHandler(console_handler)

    # 文件输出处理器
    file_handler = RotatingFileHandler(
        "logs/student-management-system.log",
        maxBytes=1024 * 1024 * 100,  # 100MB
        backupCount=10
    )
    file_handler.setLevel(logging.ERROR)
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)

    return logger
7.3 日志使用代码示例

打开app/services/users_service.py文件,修改以下代码:

from sqlalchemy.orm import Session
from app.models.users import User
from app.schemas.users import UserCreate, UserUpdate
from app.core.security import get_password_hash, verify_password
from app.utils.logger import setup_logger

logger = setup_logger(__name__)

def get_user_by_id(db: Session, user_id: int):
    try:
        user = db.query(User).filter(User.id == user_id).first()
        logger.info(f"查询用户信息成功:user_id={user_id}")
        return user
    except Exception as e:
        logger.error(f"查询用户信息失败:user_id={user_id},error={e}")
        return None

def get_user_by_username(db: Session, username: str):
    try:
        user = db.query(User).filter(User.username == username).first()
        logger.info(f"查询用户信息成功:username={username}")
        return user
    except Exception as e:
        logger.error(f"查询用户信息失败:username={username},error={e}")
        return None

def get_user_by_email(db: Session, email: str):
    try:
        user = db.query(User).filter(User.email == email).first()
        logger.info(f"查询用户信息成功:email={email}")
        return user
    except Exception as e:
        logger.error(f"查询用户信息失败:email={email},error={e}")
        return None

def get_all_users(db: Session, page: int = 1, page_size: int = 10, is_active: bool = None):
    try:
        query = db.query(User)
        if is_active is not None:
            query = query.filter(User.is_active == is_active)
        
        start_index = (page - 1) * page_size
        end_index = start_index + page_size
        users = query.offset(start_index).limit(page_size).all()
        logger.info(f"查询用户列表成功:page={page},page_size={page_size},is_active={is_active}")
        return users
    except Exception as e:
        logger.error(f"查询用户列表失败:page={page},page_size={page_size},is_active={is_active},error={e}")
        return []

def create_user(db: Session, user: UserCreate):
    try:
        hashed_password = get_password_hash(user.password)
        db_user = User(
            username=user.username,
            email=user.email,
            hashed_password=hashed_password
        )
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        logger.info(f"创建用户成功:username={user.username}")
        return db_user
    except Exception as e:
        logger.error(f"创建用户失败:username={user.username},error={e}")
        return None

def update_user(db: Session, user_id: int, user_update: UserUpdate):
    try:
        db_user = get_user_by_id(db, user_id)
        if not db_user:
            logger.warning(f"用户不存在:user_id={user_id}")
            return None
        
        if user_update.username:
            db_user.username = user_update.username
        if user_update.email:
            db_user.email = user_update.email
        if user_update.password:
            db_user.hashed_password = get_password_hash(user_update.password)
        if user_update.is_active is not None:
            db_user.is_active = user_update.is_active
        
        db.commit()
        db.refresh(db_user)
        logger.info(f"修改用户信息成功:user_id={user_id}")
        return db_user
    except Exception as e:
        logger.error(f"修改用户信息失败:user_id={user_id},error={e}")
        return None

def delete_user(db: Session, user_id: int):
    try:
        db_user = get_user_by_id(db, user_id)
        if db_user:
            db.delete(db_user)
            db.commit()
            logger.info(f"删除用户成功:user_id={user_id}")
        else:
            logger.warning(f"用户不存在:user_id={user_id}")
    except Exception as e:
        logger.error(f"删除用户失败:user_id={user_id},error={e}")

八、测试结构

测试结构是项目的重要组成部分,它决定了项目的可测试性和可维护性。测试结构的目标是覆盖项目的所有业务逻辑,确保项目的功能正常运行。

8.1 测试类型

常见的 Python 测试类型有以下几种:

  • 单元测试:测试项目的单个模块或函数;
  • 集成测试:测试项目的多个模块或函数之间的交互;
  • 系统测试:测试项目的整个系统;
  • 端到端测试:测试项目的用户界面和后端接口之间的交互。
8.2 测试框架

常见的 Python 测试框架有以下几种:

  • pytest:功能强大,易于使用,支持插件扩展;
  • unittest:Python 内置,无需安装额外依赖库;
  • doctest:Python 内置,用于测试文档字符串中的代码示例。
8.3 测试代码示例

打开conftest.py文件,输入以下代码:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.database import Base, get_db
from app.main import app

# 测试数据库的连接URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# 初始化测试数据库的引擎
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# 初始化测试数据库的会话
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建测试数据库的所有表
Base.metadata.create_all(bind=engine)

# 覆盖FastAPI的依赖注入函数
def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

# 创建FastAPI的测试客户端
client = TestClient(app)

# 测试数据
test_user_data = {
    "username": "testuser",
    "email": "testuser@example.com",
    "password": "Test1234"
}

test_student_data = {
    "student_no": "20250001",
    "name": "张三",
    "gender": "男",
    "birthdate": "2005-01-01",
    "phone": "13800138001",
    "email": "zhangsan@example.com",
    "address": "北京市海淀区"
}

打开test_auth.py文件,输入以下代码:

import pytest
from fastapi.testclient import TestClient
from app.main import app
from conftest import client, test_user_data

def test_register_user():
    """测试用户注册接口"""
    response = client.post(
        "/auth/register",
        json=test_user_data
    )
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == test_user_data["username"]
    assert data["email"] == test_user_data["email"]
    assert data["is_active"] == True

def test_register_duplicate_username():
    """测试注册重复用户名的用户"""
    response = client.post(
        "/auth/register",
        json=test_user_data
    )
    assert response.status_code == 400
    assert "用户名已存在" in response.json()["detail"]

def test_register_duplicate_email():
    """测试注册重复邮箱的用户"""
    new_user_data = {
        "username": "testuser2",
        "email": test_user_data["email"],
        "password": "Test1234"
    }
    response = client.post(
        "/auth/register",
        json=new_user_data
    )
    assert response.status_code == 400
    assert "邮箱已存在" in response.json()["detail"]

def test_login_user():
    """测试用户登录接口"""
    response = client.post(
        "/auth/login",
        data={
            "username": test_user_data["username"],
            "password": test_user_data["password"]
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"}
    )
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"

def test_login_invalid_username():
    """测试使用无效用户名登录"""
    response = client.post(
        "/auth/login",
        data={
            "username": "invaliduser",
            "password": test_user_data["password"]
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"}
    )
    assert response.status_code == 401
    assert "用户名或密码错误" in response.json()["detail"]

def test_login_invalid_password():
    """测试使用无效密码登录"""
    response = client.post(
        "/auth/login",
        data={
            "username": test_user_data["username"],
            "password": "invalidpassword"
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"}
    )
    assert response.status_code == 401
    assert "用户名或密码错误" in response.json()["detail"]

九、部署结构

部署结构是项目的重要组成部分,它决定了项目的可部署性和可维护性。部署结构的目标是将项目部署到生产环境中,确保项目的功能正常运行。

9.1 部署方式

常见的 Python 项目部署方式有以下几种:

  • 本地部署:部署到本地服务器上;
  • 云部署:部署到云平台上(如 AWS、GCP、Azure、阿里云、腾讯云);
  • 容器化部署:使用 Docker 容器化部署。
9.2 容器化部署代码示例

打开docker-compose.yml文件,输入以下代码:

version: "3.8"

services:
  mysql:
    image: mysql:8.0
    container_name: mysql
    restart: always
    environment:
      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
      MYSQL_DATABASE: ${MYSQL_DATABASE}
    ports:
      - "3306:3306"
    volumes:
      - mysql-data:/var/lib/mysql

  fastapi-app:
    build:
      context: .
      dockerfile: Dockerfile.fastapi
    container_name: fastapi-app
    restart: always
    environment:
      DATABASE_URL: ${DATABASE_URL}
      SECRET_KEY: ${SECRET_KEY}
    ports:
      - "8000:8000"
    depends_on:
      - mysql

  nginx:
    build:
      context: .
      dockerfile: Dockerfile.nginx
    container_name: nginx
    restart: always
    ports:
      - "80:80"
    depends_on:
      - fastapi-app

volumes:
  mysql-data:

打开Dockerfile.fastapi文件,输入以下代码:

FROM python:3.11-slim

WORKDIR /app

# 安装项目所需的依赖库
COPY requirements.prod.txt .
RUN pip install --no-cache-dir -r requirements.prod.txt

# 复制项目代码
COPY app /app/app
COPY database.py /app/database.py
COPY models.py /app/models.py
COPY schemas.py /app/schemas.py
COPY utils.py /app/utils.py
COPY main.py /app/main.py

# 创建logs文件夹
RUN mkdir -p /app/logs

# 暴露端口
EXPOSE 8000

# 启动FastAPI应用
CMD ["gunicorn", "app.main:app", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

打开Dockerfile.nginx文件,输入以下代码:

FROM nginx:1.24.0

# 复制Nginx配置文件
COPY nginx.conf /etc/nginx/nginx.conf

# 暴露端口
EXPOSE 80

# 启动Nginx
CMD ["nginx", "-g", "daemon off;"]

打开nginx.conf文件,输入以下代码:

events {
    worker_connections 1024;
}

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

    access_log  /var/log/nginx/access.log  main;

    sendfile        on;
    #tcp_nopush     on;

    keepalive_timeout  65;

    #gzip  on;

    server {
        listen 80;
        server_name localhost;

        location / {
            proxy_pass http://fastapi-app:8000;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
        }

        location /docs {
            proxy_pass http://fastapi-app:8000/docs;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
        }

        location /redoc {
            proxy_pass http://fastapi-app:8000/redoc;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
        }
    }
}

十、常见问题与解决方案

10.1 路径循环导入

问题:在项目开发过程中,经常会遇到路径循环导入的问题,比如app/api/endpoints/users.py导入app/services/users_service.pyapp/services/users_service.py导入app/models/users.pyapp/models/users.py导入app/core/database.pyapp/core/database.py导入app/core/config.pyapp/core/config.py导入app/models/users.py

解决方案:使用延迟导入(Lazy Import)或重构代码,避免循环导入。

10.2 数据库连接池配置

问题:在项目运行过程中,经常会遇到数据库连接池耗尽的问题,比如并发用户数过多,导致数据库连接池耗尽,API 接口响应时间过长。

解决方案:调整 SQLAlchemy 的连接池配置,增加连接池的大小和超时时间。

打开app/core/database.py文件,修改以下代码:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

engine = create_engine(
    settings.DATABASE_URL,
    pool_size=50,
    max_overflow=100,
    pool_timeout=30,
    pool_recycle=1800
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
10.3 缓存配置

问题:在项目运行过程中,经常会遇到 API 接口响应时间过长的问题,比如查询课程信息的接口需要查询数据库,每次查询都需要花费很长时间。

解决方案:使用 Redis 缓存,缓存常用的数据,减少数据库的查询次数。

10.4 CORS 配置

问题:在项目运行过程中,经常会遇到跨域资源共享(CORS)的问题,比如前端应用部署在http://localhost:3000,后端接口部署在http://localhost:8000,前端应用无法调用后端接口。

解决方案:在 FastAPI 应用中配置 CORS。

打开app/main.py文件,修改以下代码:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.endpoints import auth, users, students, teachers, courses, classes, enrollments, grades, roles
from app.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.PROJECT_VERSION,
    description=settings.PROJECT_DESCRIPTION
)

# CORS配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册API接口
app.include_router(auth.router, prefix="/auth", tags=["Auth"])
app.include_router(users.router, prefix="/users", tags=["Users"])
app.include_router(students.router, prefix="/students", tags=["Students"])
app.include_router(teachers.router, prefix="/teachers", tags=["Teachers"])
app.include_router(courses.router, prefix="/courses", tags=["Courses"])
app.include_router(classes.router, prefix="/classes", tags=["Classes"])
app.include_router(enrollments.router, prefix="/enrollments", tags=["Enrollments"])
app.include_router(grades.router, prefix="/grades", tags=["Grades"])
app.include_router(roles.router, prefix="/roles", tags=["Roles"])

十一、进阶扩展

11.1 权限控制

功能需求:用户角色管理、用户权限分配、用户角色查询、用户权限查询。

技术选型:FastAPI 的依赖注入系统、Pydantic 数据验证模型、SQLAlchemy ORM 框架。

代码示例:打开app/models/roles.py文件,输入以下代码:

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.core.database import Base

class Role(Base):
    __tablename__ = "roles"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(20), unique=True, index=True, nullable=False)
    description = Column(String(200), nullable=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    users = relationship("UserRole", back_populates="role")
    permissions = relationship("RolePermission", back_populates="role")

class Permission(Base):
    __tablename__ = "permissions"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(20), unique=True, index=True, nullable=False)
    description = Column(String(200), nullable=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    roles = relationship("RolePermission", back_populates="permission")

class UserRole(Base):
    __tablename__ = "user_roles"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    role_id = Column(Integer, ForeignKey("roles.id"), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    user = relationship("User", back_populates="roles")
    role = relationship("Role", back_populates="users")

class RolePermission(Base):
    __tablename__ = "role_permissions"

    id = Column(Integer, primary_key=True, index=True)
    role_id = Column(Integer, ForeignKey("roles.id"), nullable=False)
    permission_id = Column(Integer, ForeignKey("permissions.id"), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    role = relationship("Role", back_populates="permissions")
    permission = relationship("Permission", back_populates="roles")
11.2 API 文档优化

功能需求:API 文档的标题、版本、描述、标签、接口参数说明、接口返回值说明。

技术选型:FastAPI 的 OpenAPI 文档配置、Pydantic 数据验证模型的文档字符串。

代码示例:打开app/main.py文件,修改以下代码:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.endpoints import auth, users, students, teachers, courses, classes, enrollments, grades, roles
from app.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.PROJECT_VERSION,
    description=settings.PROJECT_DESCRIPTION,
    openapi_tags=[
        {
            "name": "Auth",
            "description": "身份验证接口"
        },
        {
            "name": "Users",
            "description": "用户管理接口"
        },
        {
            "name": "Students",
            "description": "学生管理接口"
        },
        {
            "name": "Teachers",
            "description": "教师管理接口"
        },
        {
            "name": "Courses",
            "description": "课程管理接口"
        },
        {
            "name": "Classes",
            "description": "班级管理接口"
        },
        {
            "name": "Enrollments",
            "description": "选课管理接口"
        },
        {
            "name": "Grades",
            "description": "成绩管理接口"
        },
        {
            "name": "Roles",
            "description": "权限管理接口"
        }
    ]
)

# CORS配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册API接口
app.include_router(auth.router, prefix="/auth", tags=["Auth"])
app.include_router(users.router, prefix="/users", tags=["Users"])
app.include_router(students.router, prefix="/students", tags=["Students"])
app.include_router(teachers.router, prefix="/teachers", tags=["Teachers"])
app.include_router(courses.router, prefix="/courses", tags=["Courses"])
app.include_router(classes.router, prefix="/classes", tags=["Classes"])
app.include_router(enrollments.router, prefix="/enrollments", tags=["Enrollments"])
app.include_router(grades.router, prefix="/grades", tags=["Grades"])
app.include_router(roles.router, prefix="/roles", tags=["Roles"])

十二、总结与展望

12.1 总结

通过以上步骤,我们从 0 到 1 搭了一个能维护 4 年的 Python 学生管理系统,项目采用了分层架构,代码清晰易读,功能完整,维护简单。项目的核心功能包括用户管理、学生管理、教师管理、课程管理、班级管理、选课管理、成绩管理、权限管理等。项目的技术栈包括 FastAPI、Uvicorn、Gunicorn、MySQL、SQLAlchemy、Alembic、Python-multipart、Python-jose [cryptography]、Passlib [bcrypt]、Pytest、Pytest-cov、Docker、Docker Compose、Nginx。

12.2 展望

未来,我们可以对项目进行以下优化:

  • 添加前端应用:使用 React 或 Vue 开发前端应用,提供用户友好的界面;
  • 添加 WebSocket 支持:实现实时通信功能,比如聊天功能、通知功能;
  • 添加缓存支持:使用 Redis 缓存,减少数据库的查询次数;
  • 添加消息队列支持:使用 Celery 或 RabbitMQ 实现异步任务,比如发送邮件、发送短信;
  • 添加监控支持:使用 Prometheus 和 Grafana 监控项目的运行状态;
  • 添加日志分析支持:使用 ELK Stack(Elasticsearch、Logstash、Kibana)分析项目的日志。
Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐