Chuyển đến nội dung chính

第 4 課:路徑操作、請求與回應

路徑參數、查詢參數、請求內文、標頭、Cookie。 HTTP 方法、狀態碼、回應模型、JSONResponse。使用 Pydantic 自動進行類型驗證。

💻 程式設計 — 第 4 課 第 4 課:路徑操作、請求與回應

Python FastAPI:從基礎到進階

第 1 部分:Python 基礎和 FastAPI

亞洲開發網

1. 路徑參數

路徑參數允許從 URL 路徑捕獲值。 FastAPI 會根據類型提示自動解析和驗證:

from fastapi import FastAPI, Path
from enum import Enum

app = FastAPI()

# Basic path parameter - tự động validate là int
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}
# GET /users/42 → {"user_id": 42}
# GET /users/abc → 422 Validation Error

# Multiple path parameters
@app.get("/users/{user_id}/posts/{post_id}")
async def get_user_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

# Path parameter validation với Path()
@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(
        ...,
        title="Item ID",
        description="The unique identifier of the item",
        gt=0,        # greater than 0
        le=10000,    # less than or equal to 10000
        examples=[42],
    )
):
    return {"item_id": item_id}

# Enum path parameter
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    return {"model": model_name, "value": model_name.value}
# GET /models/alexnet → {"model": "alexnet", "value": "alexnet"}
# GET /models/invalid → 422 Validation Error

# File path parameter
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}
# GET /files/home/user/data.csv → {"file_path": "home/user/data.csv"}

2. 查詢參數

查詢參數為標記後的參數 ? 在網址中:

from fastapi import FastAPI, Query

app = FastAPI()

# Basic query parameters
@app.get("/items/")
async def list_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}
# GET /items/?skip=0&limit=20

# Optional query parameter
@app.get("/search/")
async def search(q: str | None = None):
    if q:
        return {"results": f"Searching for: {q}"}
    return {"results": "No search query"}

# Required query parameter (không có default value)
@app.get("/search/required")
async def search_required(q: str):
    return {"results": f"Searching for: {q}"}
# GET /search/required → 422 Error (q is required)

# Query parameter validation với Query()
@app.get("/items/search")
async def search_items(
    q: str = Query(
        ...,
        min_length=3,
        max_length=50,
        pattern=r"^[a-zA-Z0-9\s]+$",
        title="Search Query",
        description="Search term for items",
        examples=["laptop"],
    ),
    category: str | None = Query(None, max_length=20),
    min_price: float = Query(0, ge=0),
    max_price: float = Query(10000, le=100000),
    tags: list[str] = Query(default=[]),
):
    return {
        "q": q,
        "category": category,
        "price_range": [min_price, max_price],
        "tags": tags,
    }
# GET /items/search?q=laptop&tags=gaming&tags=new

# Boolean query parameters
@app.get("/items/filter")
async def filter_items(
    is_available: bool = True,
    include_deleted: bool = False,
):
    return {"is_available": is_available, "include_deleted": include_deleted}
# GET /items/filter?is_available=true&include_deleted=false
# GET /items/filter?is_available=1  ← cũng hoạt động
# GET /items/filter?is_available=yes  ← cũng hoạt động

3. 請求體

FastAPI 使用 Pydantic 模型進行請求正文驗證:

from fastapi import FastAPI, Body
from pydantic import BaseModel, Field, field_validator
from datetime import datetime

app = FastAPI()

# Pydantic model cho request body
class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str | None = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="Price must be positive")
    tax: float | None = Field(None, ge=0)
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be empty or whitespace")
        return v.strip()

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "Laptop",
                    "description": "A powerful gaming laptop",
                    "price": 1299.99,
                    "tax": 129.99,
                    "tags": ["electronics", "gaming"],
                }
            ]
        }
    }

@app.post("/items/")
async def create_item(item: ItemCreate):
    item_dict = item.model_dump()
    if item.tax:
        item_dict["total_price"] = item.price + item.tax
    return item_dict

# Multiple body parameters
class User(BaseModel):
    name: str
    email: str

class Order(BaseModel):
    item_id: int
    quantity: int = Field(..., gt=0)

@app.post("/orders/")
async def create_order(user: User, order: Order):
    return {"user": user, "order": order}
# Request body:
# {
#   "user": {"name": "Alice", "email": "[email protected]"},
#   "order": {"item_id": 1, "quantity": 2}
# }

# Body với singular values
@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: ItemCreate,
    importance: int = Body(..., gt=0, le=5),
    note: str = Body(None),
):
    return {
        "item_id": item_id,
        "item": item,
        "importance": importance,
        "note": note,
    }

4. 標頭和 Cookie

from fastapi import FastAPI, Header, Cookie

app = FastAPI()

# Headers
@app.get("/headers/")
async def read_headers(
    user_agent: str | None = Header(None),
    x_request_id: str | None = Header(None, alias="X-Request-ID"),
    accept_language: str = Header("en"),
):
    return {
        "user_agent": user_agent,
        "request_id": x_request_id,
        "language": accept_language,
    }

# Duplicate headers (list)
@app.get("/multi-headers/")
async def read_multi_headers(
    x_token: list[str] | None = Header(None),
):
    return {"x_token": x_token}

# Cookies
@app.get("/cookies/")
async def read_cookies(
    session_id: str | None = Cookie(None),
    tracking_id: str | None = Cookie(None),
):
    return {"session_id": session_id, "tracking_id": tracking_id}

5. 回應模型和狀態代碼

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse, RedirectResponse
from pydantic import BaseModel

app = FastAPI()

# Response model - filter output fields
class UserIn(BaseModel):
    name: str
    email: str
    password: str

class UserOut(BaseModel):
    name: str
    email: str
    # password is NOT included in response!

@app.post(
    "/users/",
    response_model=UserOut,
    status_code=status.HTTP_201_CREATED,
    summary="Create a new user",
    description="Create a new user with name, email and password",
    response_description="The created user (without password)",
)
async def create_user(user: UserIn):
    # Password sẽ bị loại bỏ khỏi response nhờ response_model=UserOut
    return user

# Multiple response models
class ItemSummary(BaseModel):
    name: str
    price: float

class ItemDetail(BaseModel):
    name: str
    price: float
    description: str | None
    tax: float | None
    tags: list[str]

@app.get(
    "/items/{item_id}",
    response_model=ItemDetail,
    responses={
        200: {"description": "Item found", "model": ItemDetail},
        404: {"description": "Item not found"},
    },
)
async def get_item(item_id: int):
    ...

# Custom response
@app.get("/custom-response/")
async def custom_response():
    return JSONResponse(
        status_code=200,
        content={"message": "Custom response"},
        headers={"X-Custom-Header": "custom-value"},
    )

# Redirect
@app.get("/old-path")
async def redirect():
    return RedirectResponse(url="/new-path", status_code=status.HTTP_301_MOVED_PERMANENTLY)

# Response với exclude/include
class FullItem(BaseModel):
    name: str
    description: str | None
    price: float
    tax: float | None
    internal_code: str

@app.get(
    "/items/public/{item_id}",
    response_model=FullItem,
    response_model_exclude={"internal_code"},
    response_model_exclude_unset=True,
)
async def get_public_item(item_id: int):
    return FullItem(
        name="Laptop",
        description=None,
        price=999.99,
        tax=None,
        internal_code="INTERNAL-001",
    )
    # Response: {"name": "Laptop", "price": 999.99}
    # description, tax excluded (unset), internal_code excluded explicitly

6. 表單資料和檔案上傳

from fastapi import FastAPI, Form, File, UploadFile

app = FastAPI()

# Form data
@app.post("/login/")
async def login(
    username: str = Form(...),
    password: str = Form(...),
):
    return {"username": username}

# File upload
@app.post("/upload/")
async def upload_file(
    file: UploadFile = File(..., description="File to upload"),
):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents),
    }

# Multiple files
@app.post("/upload-multiple/")
async def upload_multiple(
    files: list[UploadFile] = File(..., description="Multiple files"),
):
    return [{"filename": f.filename, "size": f.size} for f in files]

# Form + File mixed
@app.post("/submit/")
async def submit_form(
    name: str = Form(...),
    description: str = Form(None),
    file: UploadFile = File(...),
):
    return {
        "name": name,
        "description": description,
        "filename": file.filename,
    }

7. 錯誤處理

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# HTTPException
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found",
            headers={"X-Error": "Item not found"},
        )
    return items_db[item_id]

# Custom exception
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": "item_not_found",
            "message": f"Item with id {exc.item_id} was not found",
            "path": str(request.url),
        },
    )

# Override validation error format
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " → ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
        })
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"errors": errors},
    )

8. FastAPI 中的請求生命週期

Client Request
    │
    ▼
┌─────────────────────┐
│    Middleware (1)     │  ← Process request
├─────────────────────┤
│    Middleware (2)     │
├─────────────────────┤
│    Exception Handler │  ← Catch exceptions
├─────────────────────┤
│    Dependency (DI)   │  ← Resolve dependencies
├─────────────────────┤
│    Path Operation    │  ← Execute handler
├─────────────────────┤
│    Response Model    │  ← Serialize & filter
├─────────────────────┤
│    Middleware (2)     │  ← Process response
├─────────────────────┤
│    Middleware (1)     │
└─────────────────────┘
    │
    ▼
Client Response

總結

在這篇文章中,我們了解到:

  • 路徑參數:從 URL 擷取值,使用類型提示自動驗證
  • 查詢參數:可選/必需的查詢字串,使用 Query() 進行驗證
  • 請求正文:用於結構化資料、欄位驗證器的 Pydantic 模型
  • 標頭和 Cookie:讀取HTTP headers和cookies
  • 回應模型:控制輸出格式、過濾欄位、狀態碼
  • 錯誤處理:HTTPException、自訂異常、驗證錯誤

下一篇文章將介紹 Pydantic V2 - FastAPI 最強大的序列化和驗證工具。