从 0 到 1 搭能维护 4 年的 Python 学生管理系统：FastAPI 项目结构、README、解耦全流程实战

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  # 项目说明文档

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())

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")

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

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}")

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()

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()

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()

# 学生管理系统

## 项目简介
这是一个基于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

CREATE DATABASE student_management_system CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

DATABASE_URL=mysql+pymysql://root:123456@localhost/student_management_system
SECRET_KEY=your-secret-key-here

alembic upgrade head

uvicorn app.main:app --reload

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

docker-compose up -d

docker exec -it fastapi-app alembic upgrade head

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

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

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

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

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

DATABASE_URL=mysql+pymysql://root:123456@localhost/student_management_system
SECRET_KEY=your-secret-key-here

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  # 项目说明文档

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

---

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

#### 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`文件，输入以下代码：

打开`requirements.prod.txt`文件，输入以下代码：

---

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

#### 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

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}")

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": "北京市海淀区"
}

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"]

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:

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"]

FROM nginx:1.24.0

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

# 暴露端口
EXPOSE 80

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

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;
        }
    }
}

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()

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"])

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")

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"])