跳至主要內容
LinPolly
文章 程式開發

FastAPI 快速建置 REST API：從入門到實戰

FastAPI 快速建置 REST API：從入門到實戰

FastAPI 是近年來 Python 後端開發社群最熱門的框架之一。它以 Python 型別提示為基礎，自動生成 OpenAPI 文件，效能比 Flask 和 Django REST Framework 都高出許多，同時又保持了 Python 一貫的開發效率。

這篇文章帶你從安裝到建置一個完整的 REST API，包含認證、資料庫整合和錯誤處理。

為什麼選擇 FastAPI？

  • 自動文件：基於 OpenAPI/Swagger，零額外設定
  • 型別安全：利用 Pydantic 做資料驗證
  • 高效能：基於 Starlette，效能媲美 Node.js
  • async 支援：原生支援非同步處理
  • 依賴注入：內建強大的 DI 系統

安裝與基本設定

# 建立虛擬環境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安裝依賴
pip install fastapi uvicorn[standard] pydantic sqlalchemy python-jose[cryptography] passlib[bcrypt]

第一個 API

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="我的 API",
    description="這是一個用 FastAPI 建置的 REST API",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Hello, FastAPI!"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

啟動伺服器：

uvicorn main:app --reload

開啟 http://localhost:8000/docs 就能看到自動生成的 Swagger UI。

Pydantic 資料模型

from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime
    is_active: bool

    model_config = {"from_attributes": True}  # 允許從 ORM 物件建立

class ArticleCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    content: str
    category_id: Optional[int] = None
    tags: list[str] = []

class ArticleUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None
    is_published: Optional[bool] = None

路由組織

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .. import models, schemas
from ..database import get_db

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}}
)

@router.get("/", response_model=list[schemas.UserResponse])
def get_users(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users

@router.get("/{user_id}", response_model=schemas.UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"找不到 ID 為 {user_id} 的使用者"
        )
    return user

@router.post("/", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # 檢查 email 是否已存在
    existing = db.query(models.User).filter(
        models.User.email == user.email
    ).first()
    if existing:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="此 Email 已被使用"
        )

    hashed_password = hash_password(user.password)
    db_user = models.User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

依賴注入

FastAPI 的依賴注入系統非常強大：

# dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from .database import SessionLocal
from .config import settings

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="無法驗證憑證",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise credentials_exception
    return user

async def require_admin(
    current_user = Depends(get_current_user)
):
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="需要管理員權限"
        )
    return current_user

使用依賴：

@router.delete("/{article_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_article(
    article_id: int,
    db: Session = Depends(get_db),
    current_user = Depends(require_admin)  # 只有管理員能刪除
):
    article = db.query(Article).filter(Article.id == article_id).first()
    if not article:
        raise HTTPException(status_code=404, detail="文章不存在")
    db.delete(article)
    db.commit()

全域例外處理

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc)}
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    # 記錄錯誤
    logger.error(f"未預期的錯誤: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "伺服器內部錯誤"}
    )

非同步資料庫操作

使用 asyncpg 和 SQLAlchemy async：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/dbname",
    echo=True
)

AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def get_async_db():
    async with AsyncSessionLocal() as session:
        yield session

@router.get("/articles")
async def get_articles(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(Article).order_by(Article.created_at.desc()))
    articles = result.scalars().all()
    return articles

整合應用

# main.py（完整版）
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .routers import users, articles, auth
from .config import settings

app = FastAPI(
    title=settings.APP_NAME,
    version=settings.APP_VERSION,
    docs_url="/docs" if settings.DEBUG else None,  # 生產環境關閉文件
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 路由
app.include_router(auth.router, prefix="/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/v1")
app.include_router(articles.router, prefix="/api/v1")

總結

FastAPI 是目前建置 Python REST API 的最佳選擇之一。它的型別系統、自動文件、依賴注入和非同步支援組合起來，讓你能快速打造出高品質的 API。搭配 Pydantic 做資料驗證，結合 SQLAlchemy 操作資料庫，你就擁有了一套完整且現代的 Python 後端技術棧。

#Python #Node.js

分享這篇文章