第3章 - 数据模型
🎯 本章目标
- 创建数据库模型(SQLAlchemy)
- 创建 Pydantic 模型
- 理解两种模型的区别和用途
1️⃣ 数据库模型 vs Pydantic 模型
| 类型 | 用途 | 库 |
|---|---|---|
| 数据库模型 | 定义数据库表结构,操作数据库 | SQLAlchemy |
| Pydantic 模型 | 数据验证,请求/响应格式 | Pydantic |
简单说:
- 数据库模型 = 数据怎么存
- Pydantic 模型 = 数据怎么传
2️⃣ 班级数据库模型
app/models/class_model.py
"""
班级数据库模型
"""
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from app.database import Base
class Class(Base):
"""班级表"""
__tablename__ = "classes"
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
name = Column(String(50), nullable=False, comment="班级名称")
grade = Column(String(20), nullable=False, comment="年级")
teacher = Column(String(50), nullable=True, comment="班主任")
description = Column(String(200), nullable=True, comment="班级描述")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间")
# 关联学生(一对多)
students = relationship("Student", back_populates="class_info")
def __repr__(self):
return f"<Class(id={self.id}, name={self.name})>"
3️⃣ 学生数据库模型
app/models/student.py
"""
学生数据库模型
"""
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from app.database import Base
class Student(Base):
"""学生表"""
__tablename__ = "students"
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
student_no = Column(String(20), unique=True, index=True, nullable=False, comment="学号")
name = Column(String(50), nullable=False, comment="姓名")
gender = Column(String(10), nullable=False, comment="性别")
age = Column(Integer, nullable=False, comment="年龄")
phone = Column(String(20), nullable=True, comment="联系电话")
address = Column(String(200), nullable=True, comment="地址")
class_id = Column(Integer, ForeignKey("classes.id"), nullable=True, comment="班级ID")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间")
# 关联班级(多对一)
class_info = relationship("Class", back_populates="students")
# 关联成绩(一对多)
scores = relationship("Score", back_populates="student", cascade="all, delete-orphan")
def __repr__(self):
return f"<Student(id={self.id}, name={self.name}, student_no={self.student_no})>"
4️⃣ 成绩数据库模型
app/models/score.py
"""
成绩数据库模型
"""
from sqlalchemy import Column, Integer, String, Float, Date, DateTime, ForeignKey
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from app.database import Base
class Score(Base):
"""成绩表"""
__tablename__ = "scores"
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
student_id = Column(Integer, ForeignKey("students.id", ondelete="CASCADE"), nullable=False, comment="学生ID")
subject = Column(String(50), nullable=False, comment="科目")
score = Column(Float, nullable=False, comment="分数")
exam_date = Column(Date, nullable=False, comment="考试日期")
exam_type = Column(String(20), nullable=True, default="期中考试", comment="考试类型")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间")
# 关联学生(多对一)
student = relationship("Student", back_populates="scores")
def __repr__(self):
return f"<Score(id={self.id}, student_id={self.student_id}, subject={self.subject}, score={self.score})>"
5️⃣ 模型汇总
app/models/init.py
"""
数据库模型汇总
"""
from app.models.class_model import Class
from app.models.student import Student
from app.models.score import Score
__all__ = ["Class", "Student", "Score"]
6️⃣ 班级 Pydantic 模型
app/schemas/class_schema.py
"""
班级 Pydantic 模型
"""
from typing import Optional, List
from datetime import datetime
from pydantic import BaseModel, Field
# ========== 基础模型 ==========
class ClassBase(BaseModel):
"""班级基础模型"""
name: str = Field(..., min_length=1, max_length=50, description="班级名称", example="一年级1班")
grade: str = Field(..., min_length=1, max_length=20, description="年级", example="一年级")
teacher: Optional[str] = Field(None, max_length=50, description="班主任", example="张老师")
description: Optional[str] = Field(None, max_length=200, description="班级描述")
# ========== 请求模型 ==========
class ClassCreate(ClassBase):
"""创建班级请求"""
pass
class ClassUpdate(BaseModel):
"""更新班级请求"""
name: Optional[str] = Field(None, min_length=1, max_length=50)
grade: Optional[str] = Field(None, min_length=1, max_length=20)
teacher: Optional[str] = Field(None, max_length=50)
description: Optional[str] = Field(None, max_length=200)
# ========== 响应模型 ==========
class ClassOut(ClassBase):
"""班级响应模型"""
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class ClassWithStudentCount(ClassOut):
"""带学生数量的班级响应"""
student_count: int = 0
7️⃣ 学生 Pydantic 模型
app/schemas/student.py
"""
学生 Pydantic 模型
"""
from typing import Optional, List
from datetime import datetime
from pydantic import BaseModel, Field
from enum import Enum
class GenderEnum(str, Enum):
"""性别枚举"""
male = "男"
female = "女"
# ========== 基础模型 ==========
class StudentBase(BaseModel):
"""学生基础模型"""
student_no: str = Field(..., min_length=1, max_length=20, description="学号", example="2024001")
name: str = Field(..., min_length=1, max_length=50, description="姓名", example="张三")
gender: GenderEnum = Field(..., description="性别")
age: int = Field(..., ge=1, le=100, description="年龄", example=18)
phone: Optional[str] = Field(None, max_length=20, description="联系电话", example="13800138000")
address: Optional[str] = Field(None, max_length=200, description="地址")
class_id: Optional[int] = Field(None, description="班级ID")
# ========== 请求模型 ==========
class StudentCreate(StudentBase):
"""创建学生请求"""
pass
class StudentUpdate(BaseModel):
"""更新学生请求"""
name: Optional[str] = Field(None, min_length=1, max_length=50)
gender: Optional[GenderEnum] = None
age: Optional[int] = Field(None, ge=1, le=100)
phone: Optional[str] = Field(None, max_length=20)
address: Optional[str] = Field(None, max_length=200)
class_id: Optional[int] = None
# ========== 响应模型 ==========
class StudentOut(StudentBase):
"""学生响应模型"""
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class StudentWithClass(StudentOut):
"""带班级信息的学生响应"""
class_name: Optional[str] = None
class StudentDetail(StudentOut):
"""学生详情(包含成绩)"""
class_name: Optional[str] = None
scores: List["ScoreOut"] = []
# 避免循环导入
from app.schemas.score import ScoreOut
StudentDetail.model_rebuild()
8️⃣ 成绩 Pydantic 模型
app/schemas/score.py
"""
成绩 Pydantic 模型
"""
from typing import Optional
from datetime import datetime, date
from pydantic import BaseModel, Field
from enum import Enum
class SubjectEnum(str, Enum):
"""科目枚举"""
chinese = "语文"
math = "数学"
english = "英语"
physics = "物理"
chemistry = "化学"
biology = "生物"
history = "历史"
geography = "地理"
politics = "政治"
class ExamTypeEnum(str, Enum):
"""考试类型枚举"""
midterm = "期中考试"
final = "期末考试"
monthly = "月考"
quiz = "小测"
# ========== 基础模型 ==========
class ScoreBase(BaseModel):
"""成绩基础模型"""
student_id: int = Field(..., description="学生ID")
subject: SubjectEnum = Field(..., description="科目")
score: float = Field(..., ge=0, le=100, description="分数", example=85.5)
exam_date: date = Field(..., description="考试日期")
exam_type: ExamTypeEnum = Field(default=ExamTypeEnum.midterm, description="考试类型")
# ========== 请求模型 ==========
class ScoreCreate(ScoreBase):
"""创建成绩请求"""
pass
class ScoreUpdate(BaseModel):
"""更新成绩请求"""
subject: Optional[SubjectEnum] = None
score: Optional[float] = Field(None, ge=0, le=100)
exam_date: Optional[date] = None
exam_type: Optional[ExamTypeEnum] = None
class ScoreBatchCreate(BaseModel):
"""批量创建成绩请求"""
student_id: int
scores: list[dict] # [{"subject": "语文", "score": 85}, ...]
exam_date: date
exam_type: ExamTypeEnum = ExamTypeEnum.midterm
# ========== 响应模型 ==========
class ScoreOut(ScoreBase):
"""成绩响应模型"""
id: int
created_at: datetime
class Config:
from_attributes = True
class ScoreWithStudent(ScoreOut):
"""带学生信息的成绩响应"""
student_name: str
student_no: str
9️⃣ 模型汇总
app/schemas/init.py
"""
Pydantic 模型汇总
"""
from app.schemas.class_schema import (
ClassCreate, ClassUpdate, ClassOut, ClassWithStudentCount
)
from app.schemas.student import (
StudentCreate, StudentUpdate, StudentOut,
StudentWithClass, StudentDetail, GenderEnum
)
from app.schemas.score import (
ScoreCreate, ScoreUpdate, ScoreOut,
ScoreWithStudent, ScoreBatchCreate,
SubjectEnum, ExamTypeEnum
)
__all__ = [
# 班级
"ClassCreate", "ClassUpdate", "ClassOut", "ClassWithStudentCount",
# 学生
"StudentCreate", "StudentUpdate", "StudentOut",
"StudentWithClass", "StudentDetail", "GenderEnum",
# 成绩
"ScoreCreate", "ScoreUpdate", "ScoreOut",
"ScoreWithStudent", "ScoreBatchCreate",
"SubjectEnum", "ExamTypeEnum",
]
🔟 更新主程序
更新 app/main.py,导入模型并创建表:
"""
主程序入口
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.database import engine, Base
# 导入所有模型(确保表被创建)
from app.models import Class, Student, Score
# 创建数据库表
Base.metadata.create_all(bind=engine)
# 创建 FastAPI 应用
app = FastAPI(
title=settings.PROJECT_NAME,
description=settings.PROJECT_DESCRIPTION,
version=settings.PROJECT_VERSION,
docs_url="/docs",
redoc_url="/redoc"
)
# 配置 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
def root():
return {
"message": "欢迎使用学生管理系统",
"docs": "/docs",
"redoc": "/redoc"
}
@app.get("/health")
def health_check():
return {"status": "healthy"}
📝 小结
本章我们完成了:
- ✅ 创建班级、学生、成绩的数据库模型
- ✅ 定义模型之间的关联关系
- ✅ 创建对应的 Pydantic 模型
- ✅ 区分请求模型和响应模型
🏃 下一步
模型定义好了,接下来实现 CRUD 接口!
