扫描二维码关注或者微信搜一搜：编程智域前端至全栈交流与成长

一、Tortoise-ORM模型定义

我们首先在models.py中定义评论模型：

from tortoise.models import Model
from tortoise import fields


class Comment(Model):
    id = fields.IntField(pk=True)
    content = fields.TextField()
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

    # 外键关系
    user = fields.ForeignKeyField('models.User', related_name='comments')
    article = fields.ForeignKeyField('models.Article', related_name='comments')

    class Meta:
        table = "comments"
        indexes = ("created_at", "user_id", "article_id")

    def __str__(self):
        return f"Comment {self.id} by {self.user.username}"

代码解析：

auto_now_add会在创建时自动记录时间
通过related_name建立双向关联查询路径
复合索引提升常用查询条件的效率
继承Model基类获得ORM能力

二、Pydantic模型定义

在schemas.py中定义数据验证模型：

from pydantic import BaseModel
from datetime import datetime


class CommentBase(BaseModel):
    content: str
    user_id: int
    article_id: int


class CommentCreate(CommentBase):
    pass


class CommentUpdate(BaseModel):
    content: str


class CommentResponse(CommentBase):
    id: int
    created_at: datetime
    updated_at: datetime

    class Config:
        orm_mode = True

验证要点：

创建模型继承自基础模型
更新模型仅允许修改内容字段
响应模型启用orm_mode以兼容ORM对象
时间字段自动转换时间格式

三、路由层实现

核心路由实现在comments.py中：

from fastapi import APIRouter, Depends, HTTPException
from .models import Comment
from .schemas import CommentCreate, CommentResponse

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


@router.post("/", response_model=CommentResponse)
async def create_comment(comment: CommentCreate):
    try:
        comment_obj = await Comment.create(**comment.dict())
        return await CommentResponse.from_tortoise_orm(comment_obj)
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail=f"创建评论失败: {str(e)}"
        )


@router.get("/{comment_id}", response_model=CommentResponse)
async def get_comment(comment_id: int):
    comment = await Comment.get_or_none(id=comment_id)
    if not comment:
        raise HTTPException(status_code=404, detail="评论不存在")
    return comment


@router.delete("/{comment_id}")
async def delete_comment(comment_id: int):
    deleted_count = await Comment.filter(id=comment_id).delete()
    if not deleted_count:
        raise HTTPException(status_code=404, detail="评论不存在")
    return {"message": "评论删除成功"}

技术要点：

使用get_or_none替代get避免直接抛出异常
批量删除返回影响行数作为判断依据
异常处理覆盖数据库操作的各种失败场景

四、测试接口

使用requests测试接口：

import requests

BASE_URL = "http://localhost:8000/comments"


# 创建测试
def test_create_comment():
    data = {
        "content": "优质技术文章！",
        "user_id": 1,
        "article_id": 1
    }
    response = requests.post(BASE_URL, json=data)
    assert response.status_code == 200
    print(response.json())


# 异常测试
def test_invalid_user():
    data = {
        "content": "错误测试",
        "user_id": 999,
        "article_id": 1
    }
    response = requests.post(BASE_URL, json=data)
    assert response.status_code == 400
    print(response.json())

五、课后Quiz

当查询不存在的评论ID时，应该返回什么HTTP状态码？
A) 200
B) 404
C) 500
D) 400

答案：B) 404。get_or_none方法会返回None，触发自定义的404异常

如何实现评论的软删除功能？
A) 直接删除数据库记录
B) 添加is_deleted字段
C) 使用数据库回收站功能
D) 修改内容为”已删除”

答案：B) 添加布尔型is_deleted字段，查询时过滤已删除的记录

六、常见报错处理

报错：tortoise.exceptions.IntegrityError: FOREIGN KEY constraint failed
原因：尝试关联不存在的用户或文章ID
解决：检查外键值是否存在，添加数据库约束
报错：pydantic.error_wrappers.ValidationError
原因：请求体缺少必填字段或字段类型错误
解决：检查请求体是否匹配schema定义，使用try-except捕获验证错误
报错：RuntimeError: Event loop is closed
原因：异步操作未正确关闭
解决：在main.py中添加关闭事件循环的hook：

from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()

register_tortoise(
    app,
    db_url="sqlite://db.sqlite3",
    modules={"models": ["app.models"]},
    generate_schemas=True,
    add_exception_handlers=True,
)

余下文章内容请点击跳转至个人博客页面或者扫码关注或者微信搜一搜：编程智域前端至全栈交流与成长，阅读完整的文章：

使用Tortoise-ORM和FastAPI构建评论系统