FastAPI 实战：打造高效的 CRUD 接口（增删改查优秀实践）
本文你将学到：
如何编写标准的增删改查接口 如何返回统一格式的响应如何处理异常与不存在的数据如何编写清晰可维护的代码结构
复制
app/
├── api/
│ └── user.py # 用户接口
├── models/
│ └── user.py # ORM 数据模型
├── schemas/
│ └── user.py # Pydantic 请求 / 响应模型
├── core/
│ └── db.py # 数据库初始化
├── main.py # 项目入口1.2.3.4.5.6.7.8.9.10.
复制
from tortoise.models import Model
from tortoise import fields
class User(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=50)
email = fields.CharField(max_length=100, unique=True)
class Meta:
table = "user"1.2.3.4.5.6.7.8.9.10.
复制
from pydantic import BaseModel, EmailStr
classUserCreate(BaseModel):
name: str
email: EmailStr
classUserUpdate(BaseModel):
name: str | None = None
email: EmailStr | None = None
classUserOut(BaseModel):
id: int
name: str
email: EmailStr
classConfig:
orm_mode = True1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.
复制
from fastapi import APIRouter, HTTPException
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate, UserOut
router = APIRouter(prefix="/users", tags=["用户"])
# 🔹 创建用户
@router.post("", response_model=UserOut)
asyncdefcreate_user(user: UserCreate):
exists = await User.get_or_none(email=user.email)
if exists:
raise HTTPException(status_code=400, detail="邮箱已存在")
user_obj = await User.create(**user.dict())
return user_obj
# 🔹 获取单个用户
@router.get("/{user_id}", response_model=UserOut)
asyncdefget_user(user_id: int):
user = await User.get_or_none(id=user_id)
ifnot user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
# 🔹 获取所有用户
@router.get("", response_model=list[UserOut])
asyncdeflist_users():
returnawait User.all()
# 🔹 更新用户
@router.put("/{user_id}", response_model=UserOut)
asyncdefupdate_user(user_id: int, user: UserUpdate):
user_obj = await User.get_or_none(id=user_id)
ifnot user_obj:
raise HTTPException(status_code=404, detail="用户不存在")
user_data = user.dict(exclude_unset=True)
for k, v in user_data.items():
setattr(user_obj, k, v)
await user_obj.save()
return user_obj
# 🔹 删除用户
@router.delete("/{user_id}")
asyncdefdelete_user(user_id: int):
deleted_count = await User.filter(id=user_id).delete()
ifnot deleted_count:
raise HTTPException(status_code=404, detail="用户不存在")
return {"message": "删除成功"}1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.24.25.26.27.28.29.30.31.32.33.34.35.36.37.38.39.40.41.42.43.44.45.46.47.
复制
from tortoise.contrib.fastapi import register_tortoise
from fastapi import FastAPI
def init_db(app: FastAPI):
register_tortoise(
app,
db_url="sqlite://db.sqlite3",
modules={"models": ["app.models.user"]},
generate_schemas=True,
add_exception_handlers=True,
)1.2.3.4.5.6.7.8.9.10.11.
复制
from fastapi import FastAPI
from app.core.db import init_db
from app.api import user
app = FastAPI(title="FastAPI CRUD 示例")
app.include_router(user.router)
init_db(app)1.2.3.4.5.6.7.8.
运行项目后访问：
复制
http://localhost:8000/docs1.
你可以直接使用内置的 Swagger UI 测试所有接口，无需手动构造请求！
总结本篇你学到了如何基于 FastAPI 和 Tortoise-ORM 构建一套标准的增删改查接口。结构清晰、代码优雅、响应规范，是后续扩展的基础！
阅读剩余
THE END
