FastAPI 實(shí)戰(zhàn):打造高效的 CRUD 接口(增刪改查優(yōu)秀實(shí)踐)
作者:Ss肥魚
快速、安全、優(yōu)雅地構(gòu)建一套完整的數(shù)據(jù)庫 API,是每個(gè)全棧工程師的基本功。本篇文章,我們用 FastAPI + Tortoise-ORM 實(shí)現(xiàn)最常見的 用戶信息表的 CRUD 接口。
本文你將學(xué)到:
- 如何編寫標(biāo)準(zhǔn)的增刪改查接口
- 如何返回統(tǒng)一格式的響應(yīng)
- 如何處理異常與不存在的數(shù)據(jù)
- 如何編寫清晰可維護(hù)的代碼結(jié)構(gòu)
1. 項(xiàng)目結(jié)構(gòu)一覽(適用于多模塊項(xiàng)目)
app/
├── api/
│ └── user.py # 用戶接口
├── models/
│ └── user.py # ORM 數(shù)據(jù)模型
├── schemas/
│ └── user.py # Pydantic 請求 / 響應(yīng)模型
├── core/
│ └── db.py # 數(shù)據(jù)庫初始化
├── main.py # 項(xiàng)目入口
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. 請求與響應(yīng)模型(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 接口實(shí)現(xiàn)(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=["用戶"])
# ?? 創(chuàng)建用戶
@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
# ?? 獲取單個(gè)用戶
@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. 注冊數(shù)據(jù)庫(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. 項(xiàng)目入口(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 自動(dòng)文檔)
運(yùn)行項(xiàng)目后訪問:
http://localhost:8000/docs
你可以直接使用內(nèi)置的 Swagger UI 測試所有接口,無需手動(dòng)構(gòu)造請求!
總結(jié)
本篇你學(xué)到了如何基于 FastAPI 和 Tortoise-ORM 構(gòu)建一套標(biāo)準(zhǔn)的增刪改查接口。結(jié)構(gòu)清晰、代碼優(yōu)雅、響應(yīng)規(guī)范,是后續(xù)擴(kuò)展的基礎(chǔ)!
責(zé)任編輯:趙寧寧
來源:
Ssoul肥魚