FastAPI 實戰:打造高效的 CRUD 接口(增刪改查優秀實踐)
作者:Ss肥魚
快速、安全、優雅地構建一套完整的數據庫 API,是每個全棧工程師的基本功。本篇文章,我們用 FastAPI + Tortoise-ORM 實現最常見的 用戶信息表的 CRUD 接口。
本文你將學到:
- 如何編寫標準的增刪改查接口
- 如何返回統一格式的響應
- 如何處理異常與不存在的數據
- 如何編寫清晰可維護的代碼結構
1. 項目結構一覽(適用于多模塊項目)
app/
├── api/
│ └── user.py # 用戶接口
├── models/
│ └── user.py # ORM 數據模型
├── schemas/
│ └── user.py # Pydantic 請求 / 響應模型
├── core/
│ └── db.py # 數據庫初始化
├── main.py # 項目入口
2. ORM 模型定義(models/user.py)
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"
3. 請求與響應模型(schemas/user.py)
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 = True
4. CRUD 接口實現(api/user.py)
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": "刪除成功"}
5. 注冊數據庫(core/db.py)
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,
)
6. 項目入口(main.py)
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)
7. 測試建議(使用 Swagger 自動文檔)
運行項目后訪問:
http://localhost:8000/docs
你可以直接使用內置的 Swagger UI 測試所有接口,無需手動構造請求!
總結
本篇你學到了如何基于 FastAPI 和 Tortoise-ORM 構建一套標準的增刪改查接口。結構清晰、代碼優雅、響應規范,是后續擴展的基礎!
責任編輯:趙寧寧
來源:
Ssoul肥魚