在项目中，数据库的设计与实体关系建模是关键环节，对整个平台的稳定运行和功能实现起着决定性作用。本次任务我对项目的需求和功能进行了进一步的分析，并完成了初步的数据库设计，并在后端实现了各个数据库对应的方法，完善了用户权限的审核。

一、数据库设计分析

对我们的AI智能辩论教练项目进行分析，按照数据库设计的思路，系统核心功能可以大体分为以下几点：

• 用户登录与画像管理

• 智能陪练与仿真辩论

• 自动评分与能力提升建议

• 观点检索与结构化语料使用

• 社区互动（话题、评论）

为支撑这些业务场景，初步设计了一套数据库结构，采用第三范式（3NF）设计数据表，确保数据不冗余、逻辑独立，并建模了以下基础实体：

• User：用户（辩手/管理员）

• DebateSession：每场辩论训练记录

• DebateTurn：一次辩论中的每一轮发言

• Report：每场辩论的评分反馈与个性化训练建议报告

• ArgumentCorpus：结构化语料库

• CommunityPost：社区发帖

• Comment：帖子评论

二、实体关系建模

根据上述数据库基础实体，构建实体之间的关系，以上实体对应的E-R图如下图所示

其中ArgumentCorpus结构化语料库由小组另一位负责RAG的同学建立，与其他实体耦合度较低，因此暂时不放在E-R图中。

三、数据库相关代码实现

在实际的工程中，主数据库使用关系型数据库MySQL，结构化存储各种数据。

接下来我将以debate_session辩论会话实体来展示在FastAPI中数据模型和接口方法的定义与实现。

1. 数据模型定义

首先明确debate_session实体中可能出现的各个属性的含义

• 记录每场辩论的主题（topic）、立场（position）与结果（result）

• 关联发起辩论的用户（user_id）

• 自动记录辩论发起时间（created_at）

基于以上设计目标，我们使用 SQLModel（结合了 Pydantic + SQLAlchemy 的现代ORM框架）定义了数据模型。

首先，在后端的models模块下创建新文件debate_session_models.py，在里面定义不同的数据模型，这种定义方式对我来说很新颖，因此我思考并查阅资料，了解了这样设计的目的。

为什么要定义不同的数据模型？

在传统 Web 开发中（比如Spring Boot + Hibernate、Django ORM），我们通常会这样设计数据模型：

• 定义一张数据库表（Entity）

• 直接用数据库表对象（比如 Java 的 Entity，Python 的 Model）来完成所有 CRUD（增删改查）操作

• 请求创建、更新、返回时，往往直接复用这一张表的数据模型

这种方式虽然简洁，但也带来几个问题：

• 不够灵活：创建和更新时字段要求不同，却只能用同一个模型

• 容易安全隐患：接口返回时容易泄露敏感字段（比如密码哈希）

• 无法精准验证：每个操作的输入输出格式很难做到严格控制

而在 SQLModel 中，它推崇一种新的实践模式：为每种不同的使用场景单独定义专属的数据模型（Pydantic Model）

比如针对一张 debate_sessions 表，它会专门为不同的用途定义响应的数据模型，例如更新模型、创建模型、返回模型、数据库表模型等，这样的好处非常明显：

• 不同场景不同模型，精准校验，避免出错

• 返回时可以控制隐藏敏感字段，提升安全性

•  易于维护，未来表结构变更时影响面小

针对这种模式，我设计了debate_session实体可能会用到的各种数据模型。其在数据库表中模型中，要定义 user_id 外键关联到 users 表，实现辩论会话与用户的绑定，同时也需要在User中定义与DebateSession的关联

# app/models/debate_models.py
import uuid
from datetime import datetime
from sqlmodel import SQLModel, Field, Relationship
from app.models.user_models import User

#基础模型，包含其他模型都应该有的属性
class DebateSessionBase(SQLModel):
    topic: str
    position: str  # pro / con
    result: str | None = None  # win / lose / draw

#创建模型
class DebateSessionCreate(DebateSessionBase):
    pass

#更新模型
class DebateSessionUpdate(SQLModel):
    topic: str | None = None
    position: str | None = None
    result: str | None = None

#数据库表模型，与实际的数据库表格属性对应
class DebateSession(DebateSessionBase, table=True):
    __tablename__ = "debate_sessions"
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    user_id: uuid.UUID = Field(foreign_key="users.id", nullable=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    user: User | None = Relationship(back_populates="debate_sessions")

#返回模型（用于API返回）
class DebateSessionPublic(DebateSessionBase):
    id: uuid.UUID
    user_id: uuid.UUID
    created_at: datetime

#返回列表模型
class DebateSessionsPublic(SQLModel):
    data: list[DebateSessionPublic]
    count: int

2. CRUD接口方法实现

（1）用户权限控制

为保证用户的信息和隐私安全，只有在确认用户身份和权限的时候，才能针对每个用户进行相应的操作。在第二篇博客中已经完成了登录的实现和token的返回，接下来需要在后面的每次请求中，都加入该用户的token，后端对token进行解码和权限认证后，方可调用相应接口并响应请求。

在deps.py中定义解析token并返回当前用户的方法get_current_user，以下是传入的参数：

• SessionDep：表示数据库会话对象Session，通过 Depends(get_db) 自动注入数据库连接。

• TokenDep：表示当前请求头中携带的Bearer Token（JWT），由Depends(reusable_oauth2) 提取

在该方法中对token进行解码并找到对应的用户，作为方法的返回值。在后面需要进行用户身份认证的方法中，只需要传入别名CurrentUser，FastAPI 就会自动获取 JWT Token、解码 token、查询用户并注入 User 实例，极大地提升了代码的可读性与复用性

SessionDep = Annotated[Session, Depends(get_db)]
TokenDep = Annotated[str, Depends(reusable_oauth2)]


def get_current_user(session: SessionDep, token: TokenDep) -> User:
    try:
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
        )
        token_data = TokenPayload(**payload)

    except (InvalidTokenError, ValidationError):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Could not validate credentials",
        )
    user = session.get(User, UUID(token_data.sub))
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    # if not user.is_active:
    #     raise HTTPException(status_code=400, detail="Inactive user")
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]

对于前端，在登录时将后端返回的token存入到了cookie中，因此在后续需要身份验证的请求中，需要把token添加到请求头中。首先，从 cookie 读取 token，在请求拦截器中加入携带token请求头Authorization。

//从cookie中获取 token
const getTokenFromCookie = () => {
  const match = document.cookie.match(/authToken=([^;]*)/);
  return match ? match[1] : null;
};

//在请求拦截器中添加token
axios.interceptors.request.use(config => {
  const token = getTokenFromCookie();
  if (token) {
    config.headers.Authorization = `Bearer ${token}`;
  }
  return config;
});

可以看到请求头已经被正确设置

（2）具体方法实现

接口文件位于 app/api/routes/debate.py，按照标准 RESTful 风格实现了对DebateSession增删改查，其中完成了对数据库会话的增删改查操作，以及对出现的错误的判断，并返回清晰的错误状态码和报错信息。

import uuid
from datetime import datetime
from typing import Any

from fastapi import APIRouter, HTTPException
from sqlmodel import func, select

from app.api.deps import CurrentUser, SessionDep
from app.models.debate_session_models import (
    DebateSession,
    DebateSessionCreate,
    DebateSessionUpdate,
    DebateSessionPublic,
    DebateSessionsPublic,
)

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


@router.get("/", response_model=DebateSessionsPublic)
def read_debate_sessions(
    session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
) -> Any:
    if current_user.is_superuser:
        count = session.exec(select(func.count()).select_from(DebateSession)).one()
        items = session.exec(select(DebateSession).offset(skip).limit(limit)).all()
    else:
        count = session.exec(
            select(func.count()).select_from(DebateSession).where(DebateSession.user_id == current_user.id)
        ).one()
        items = session.exec(
            select(DebateSession)
            .where(DebateSession.user_id == current_user.id)
            .offset(skip)
            .limit(limit)
        ).all()
    return DebateSessionsPublic(data=items, count=count)


@router.get("/{id}", response_model=DebateSessionPublic)
def read_debate_session(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any:
    debate = session.get(DebateSession, id)
    if not debate:
        raise HTTPException(status_code=404, detail="Debate session not found")
    if not current_user.is_superuser and debate.user_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not enough permissions")
    return debate


@router.post("/", response_model=DebateSessionPublic)
def create_debate_session(
    *, session: SessionDep, current_user: CurrentUser, item_in: DebateSessionCreate
) -> Any:
    debate = DebateSession.model_validate(item_in, update={"user_id": current_user.id})
    session.add(debate)
    session.commit()
    session.refresh(debate)
    return debate


@router.put("/{id}", response_model=DebateSessionPublic)
def update_debate_session(
    *, session: SessionDep, current_user: CurrentUser, id: uuid.UUID, item_in: DebateSessionUpdate
) -> Any:
    debate = session.get(DebateSession, id)
    if not debate:
        raise HTTPException(status_code=404, detail="Debate session not found")
    if debate.user_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not enough permissions")
    update_dict = item_in.model_dump(exclude_unset=True)
    debate.sqlmodel_update(update_dict)
    session.add(debate)
    session.commit()
    session.refresh(debate)
    return debate


@router.delete("/{id}")
def delete_debate_session(
    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
) -> dict:
    debate = session.get(DebateSession, id)
    if not debate:
        raise HTTPException(status_code=404, detail="Debate session not found")
    if not current_user.is_superuser and debate.user_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not enough permissions")
    session.delete(debate)
    session.commit()
    return {"message": "Debate session deleted successfully"}

在前端对接口进行测试，发现可以正确完成用户身份的验证，并返回对应的数据。