前昨天搞定了 Tortoise-ORM 的基础用法,今天项目里正式用上了模型之间的关联关系。趁着刚写完接口,把一对一和一对多(多对一)关系的增删改查思路整理一下,顺便记录几个踩过的小坑。
一、先分清这几种关系到底是什么
数据库里两张表之间的关系,说白了就三种:
- 一对一:一条 A 记录只对应一条 B 记录,反过来也一样。比如一个用户只有一份资料。
- 一对多 / 多对一:这其实是同一个关系的两个视角。比如一个分类下可以有很多篇文章(分类看文章是"一对多"),反过来每篇文章只属于一个分类(文章看分类是"多对一")。
- 多对多:两边都能对应多条记录,比如一篇文章可以打多个标签,一个标签也能挂在多篇文章上,这种关系需要一张中间表。
在项目里,User 和 User_Profiles 是一对一,Category/User 和 Article 是一对多(多对一),Article 和 Tag 是多对多。今天主要聊前两种。
二、一对一关系:User 与 User_Profiles
模型定义
一对一在 Tortoise-ORM 里用 OneToOneField 声明,只需要在"从表"里写一次:
class User_Profiles(models.Model):
id = fields.IntField(pk=True)
user = fields.OneToOneField("models.User", related_name="profile")
nickname = fields.CharField(max_length=20, null=True)
avatar_url = fields.CharField(max_length=200, null=True)
bio = fields.TextField(null=True)
phone = fields.CharField(max_length=11, null=True)
related_name="profile" 的作用是反向访问,拿到一个 User 对象后可以直接 user_obj.profile 拿到对应的资料记录,不用单独发一次查询。
查资料 / 改资料
查询接口很直接,按 ID 查一条:
@user_router.get("/info/{user_id}", summary="获取用户信息")
async def user_info(user_id: int):
userinf = await models.User_Profiles.filter(id=user_id).first()
if not userinf:
return {"code": "400", "msg": "用户不存在"}
return {"code": "200", "msg": "获取用户资料成功", "data": userinf}
修改资料用的是老熟人 exclude_unset=True + setattr() 组合,只更新前端实际传了的字段:
@user_router.put("/update/{user_id}", summary="修改用户信息")
async def user_update(user_id: int, user: user.User_Profiles_UpdateRequest):
userinfo = await models.User_Profiles.filter(id=user_id).first()
if not userinfo:
return {"code": "400", "msg": "用户不存在"}
userinfo_dict = user.dict(exclude_unset=True)
for key, value in userinfo_dict.items():
setattr(userinfo, key, value)
await userinfo.save()
return {"code": "200", "msg": "修改用户资料成功", "data": userinfo.id}
一对一关系的增删改查其实跟单表操作没什么区别,唯一的区别是模型里多了一个指向另一张表的外键字段,本身不涉及复杂的关联查询逻辑。
三、一对多 / 多对一:Category、User 与 Article
模型定义
一对多用 ForeignKeyField 声明在"多"的那一方(子表):
class Article(models.Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=100)
content = fields.TextField()
category = fields.ForeignKeyField("models.Category", related_name="articles_category")
user = fields.ForeignKeyField("models.User", related_name="articles_user")
view_count = fields.IntField(default=0)
status = fields.SmallIntField(default=0)
一篇文章必须挂在一个分类下、属于一个用户,这就是"多对一";反过来一个分类下能有很多篇文章,就是"一对多"。数据库层面对应的是外键约束 ON DELETE CASCADE,父记录被删时子记录也会跟着清空。
添加文章:先校验外键是否存在
新增关联数据时,最容易踩的坑就是外键 ID 传了个不存在的值导致数据库报错。项目里的做法是先手动查一遍,存在才继续:
@article_router.post("/add", summary="添加文章")
async def article_add(article: ArticleaddRequest):
articles = await models.Article.filter(title=article.title).first()
if articles:
return {"code": "400", "msg": "文章已存在"}
user = await models.User.filter(id=article.user_id).first()
if not user:
return {"code": "400", "msg": "用户不存在"}
category = await models.Category.filter(id=article.category_id).first()
if not category:
return {"code": "400", "msg": "分类不存在"}
article_add = await models.Article.create(
title=article.title,
content=article.content,
category_id=article.category_id,
user_id=article.user_id,
view_count=article.view_count,
status=article.status,
)
return {"code": "200", "msg": "添加成功"}
注意这里直接传的是 category_id / user_id,而不是 category / user 对象——Tortoise 允许直接用外键字段名加 _id 后缀来赋值,省去了一次额外查询再传对象的步骤。
查询:用 prefetch_related 一次性带出关联数据
查文章列表时如果每篇文章都单独去查一次分类名、作者名,很容易演变成经典的 N+1 查询问题。这里用 prefetch_related 提前把关联对象一起取出来:
@article_router.get("/all", summary="查询文章")
async def article_list(keyword: str = Query(None), category_id: int = Query(None)):
query = models.Article.all().prefetch_related("category", "user", "tags")
if keyword:
query = query.filter(title__icontains=keyword)
if category_id:
query = query.filter(category_id=category_id)
articles = await query
return {"code": "200", "msg": "查询成功", "data": [
{"id": i.id, "title": i.title, "category": i.category.name,
"user": i.user.username, "tags": [t.name for t in i.tags]}
for i in articles
]}
分类列表接口也是同样的思路,用 prefetch_related("articles_category") 把该分类下的文章一起带出来,用来统计文章数量:
categories = await models.Category.all().prefetch_related("articles_category")
# …
"articles_count": len(i.articles_category)
删除:先检查有没有子记录
分类删除接口里有个很关键的保护逻辑——不能直接删掉还挂着文章的分类:
@category_router.delete("/delete/{id}", summary="删除分类")
async def category_delete(id: int):
categorys = await models.Category.filter(id=id).first()
if not categorys:
return {"code": "400", "msg": "分类不存在"}
article = await models.Article.filter(category_id=id).exists()
if article:
return {"code": "400", "msg": "该分类下有文章,请先删除文章"}
await categorys.delete()
return {"code": "200", "msg": "删除成功"}
虽然数据库外键设置的是 ON DELETE CASCADE(父表删除会级联删子表),但业务层面往往不希望"删个分类顺手把几十篇文章都删了"这种事故发生,所以在接口里加一层业务校验会更稳妥,exists() 方法只判断存不存在、不用把整批数据都查出来,效率也更高。
四、路由拆分与统一注册
项目里按业务模块把接口拆成了 user_api、article_api、category_api、tag_api 四个路由文件,每个都用 APIRouter(prefix=…, tags=[…]) 声明自己的前缀和文档分组,最后统一在 main.py 里 include_router:
app.include_router(user_router)
app.include_router(article_router)
app.include_router(category_router)
app.include_router(tag_router)
这样接口一多也不会乱成一团,Swagger 文档里还会按 tags 自动分组展示,找接口的时候方便很多。




