使用Tortoise-ORM和FastAPI构建评论系统
title: 使用Tortoise-ORM和FastAPI构建评论系统
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}"
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
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": "评论删除成功"}
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())
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,
)
评论
发表评论