qidao123.com技术社区-IT企服评测·应用市场

标题: FastAPI与Tortoise-ORM开发的神奇之旅 [打印本页]

作者: 天津储鑫盛钢材现货供应商 时间: 2025-5-5 01:12
标题: FastAPI与Tortoise-ORM开发的神奇之旅
title: FastAPI与Tortoise-ORM开发的神奇之旅
date: 2025/05/05 00:15:48
updated: 2025/05/05 00:15:48
author: cmdragon
excerpt:
FastAPI与Tortoise-ORM联合实现全链路开发，涵盖环境设置、数据模型定义、Pydantic模式计划及API端点实现。文章模型包含软删除功能，通过is_deleted字段实现逻辑删除。API支持创建、分页查询、条件更新和软删除操作，确保数据安全与完整性。分页机制通过offset和limit实现，条件更新利用exclude_unset=True避免未传字段被覆盖。软删除通过filter().update()实现原子操作。体系具备用户认证、输入验证、防注入和细粒度权限控制等安全特性。
categories:

后端开发
FastAPI

tags:

FastAPI
TortoiseORM
全链路开发
数据模型
API端点
分页机制
软删除

扫描二维码
关注大概微信搜一搜：编程智域前端至全栈交流与成长
探索数千个预构建的 AI 应用，开启你的下一个伟大创意：https://tools.cmdragon.cn/
从API到数据库：FastAPI + Tortoise-ORM全链路开发指南

1. 环境准备与包安装

起首创建Python虚拟环境并安装必要依赖：

python -m venv venv
source venc/bin/activate # Linux/Mac
venv\Scripts\activate # Windows
pip install fastapi uvicorn tortoise-orm pydantic python-multipart

复制代码

2. 数据模型定义

创建包含软删除功能的文章模型：

# models.py
from tortoise.models import Model
from tortoise import fields
class Article(Model):
id = fields.IntField(pk=True)
title = fields.CharField(max_length=255)
content = fields.TextField()
author_id = fields.IntField()
created_at = fields.DatetimeField(auto_now_add=True)
updated_at = fields.DatetimeField(auto_now=True)
is_deleted = fields.BooleanField(default=False)
class Meta:
table = "articles"
class PydanticMeta:
exclude = ["is_deleted"]

复制代码

3. Pydantic模式定义

定义请求响应数据布局：

# schemas.py
from pydantic import BaseModel
from datetime import datetime
class ArticleCreate(BaseModel):
title: str
content: str
class ArticleUpdate(BaseModel):
title: str | None = None
content: str | None = None
class ArticleResponse(BaseModel):
id: int
title: str
content: str
author_id: int
created_at: datetime
updated_at: datetime
class Config:
orm_mode = True

复制代码

4. FastAPI应用设置

设置数据库毗连和路由：

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from tortoise.contrib.fastapi import register_tortoise
app = FastAPI()
# 数据库配置
DATABASE_URL = "sqlite://./db.sqlite3"
register_tortoise(
app,
db_url=DATABASE_URL,
modules={"models": ["models"]},
generate_schemas=True,
add_exception_handlers=True,
)
# 模拟用户认证
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(token: str = Depends(oauth2_scheme)):
return {"id": 1} # 模拟返回用户信息

复制代码

5. 完整API端点实现

5.1 创建文章

@app.post("/articles", response_model=ArticleResponse)
async def create_article(
article: ArticleCreate,
user: dict = Depends(get_current_user)
):
db_article = await Article.create(
**article.dict(),
author_id=user["id"]
)
return await ArticleResponse.from_tortoise_orm(db_article)

复制代码

5.2 分页查询

@app.get("/articles", response_model=list[ArticleResponse])
async def list_articles(
page: int = 1,
per_page: int = 10
):
skip = (page - 1) * per_page
query = Article.all().offset(skip).limit(per_page)
return await ArticleResponse.from_queryset(query)

复制代码

5.3 条件更新

@app.patch("/articles/{article_id}", response_model=ArticleResponse)
async def update_article(
article_id: int,
update_data: ArticleUpdate,
user: dict = Depends(get_current_user)
):
# 只允许作者修改自己的文章
db_article = await Article.get_or_none(id=article_id, author_id=user["id"])
if not db_article:
raise HTTPException(status_code=404, detail="Article not found")
await db_article.update_from_dict(update_data.dict(exclude_unset=True))
await db_article.save()
return await ArticleResponse.from_tortoise_orm(db_article)

复制代码

5.4 软删除

@app.delete("/articles/{article_id}")
async def delete_article(
article_id: int,
user: dict = Depends(get_current_user)
):
updated_count = await Article.filter(
id=article_id,
author_id=user["id"]
).update(is_deleted=True)
if not updated_count:
raise HTTPException(status_code=404, detail="Article not found")
return {"message": "Article deleted"}

复制代码

6. 关键实现剖析

6.1 分页机制原理

计算skip值：skip = (page - 1) * per_page
数据库查询利用offset和limit实现分页
实际生产环境建议利用游标分页（Cursor Pagination）

6.2 条件更新安全

通过get_or_none确保记录存在
利用filter().update()实现原子操作
exclude_unset=True忽略未传字段

6.3 软删除实现

添加is_deleted字段
重写默认查询过滤器：

class SoftDeleteManager(fields.Model):
def get_queryset(self):
return super().get_queryset().filter(is_deleted=False)

复制代码

7. 课后Quiz

Q1：怎样防止SQL注入攻击？
A：利用ORM的参数化查询，避免拼接SQL语句。Tortoise-ORM会自动处理查询参数，例如：

await Article.filter(title="Safe' OR 1=1--")

复制代码

实际天生的SQL会是参数化查询，特殊字符会被正确转义
Q2：为什么在更新时利用exclude_unset=True？
A：该参数会忽略未提供的字段，实现部门更新。例如用户只修改标题时，不会影响content字段
Q3：怎样实现真正的物理删除？
A：直接利用delete()方法：

await Article.filter(id=1).delete()

复制代码

8. 常见报错解决方案

错误1：422 Validation Error

原因：请求体不符合Pydantic模型要求
解决：检查请求头Content-Type是否正确（应为application/json），确认字段类型和必填项

错误2：404 Not Found

原因：尝试操作不存在的记录
解决：在更新/删除前先利用get_or_none检查记录是否存在

错误3：RuntimeError - Event loop closed

原因：在同步代码中调用异步方法
解决：确保所有数据库操作都在async函数中执行，利用await调用

启动服务下令：

uvicorn main:app --reload

复制代码

访问 http://localhost:8000/docs 即可测试API接口。本文实现的体系具备以下安全特性：

用户认证隔离数据访问
所有输入参数自动验证
数据库操作防注入
细粒度的权限控制

余下文章内容请点击跳转至个人博客页面大概扫码关注大概微信搜一搜：编程智域前端至全栈交流与成长，阅读完整的文章：FastAPI与Tortoise-ORM开发的神奇之旅 | cmdragon's Blog
往期文章归档：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/)