调试与边界：错误处理、日志、测试

错误处理

1. HTTPException

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": {"name": "Foo"}}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

测试：

curl http://localhost:8000/items/bar
# 输出: {"detail":"Item not found"}
# 状态码: 404

2. 自定义异常处理器

from fastapi import Request
from fastapi.responses import JSONResponse

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something wrong."},
    )

@app.get("/custom/{name}")
async def custom(name: str):
    if name == "error":
        raise CustomException(name=name)
    return {"name": name}

3. 验证错误处理

from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=400)

日志记录

配置日志

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    logger.info(f"Fetching item {item_id}")
    try:
        item = get_item(item_id)
        logger.info(f"Item {item_id} found")
        return item
    except Exception as e:
        logger.error(f"Error fetching item {item_id}: {str(e)}")
        raise

请求日志中间件

import time

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    logger.info(f"Request: {request.method} {request.url}")
    
    response = await call_next(request)
    
    process_time = time.time() - start_time
    logger.info(f"Response: {response.status_code} - {process_time:.2f}s")
    
    return response

边界情况

1. 空值和可选参数

from typing import Optional

@app.get("/items/")
async def read_items(
    q: Optional[str] = None,
    skip: int = 0,
    limit: int = 100
):
    # q 可以是 None
    if q:
        # 有查询参数
        items = search_items(q)
    else:
        # 无查询参数
        items = get_all_items()
    
    return items[skip:skip + limit]

2. 类型转换边界

from pydantic import validator

class Item(BaseModel):
    name: str
    price: float
    
    @validator('price')
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('价格必须大于 0')
        return v
    
    @validator('name')
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('名称不能为空')
        return v.strip()

3. 大文件上传

from fastapi import File, UploadFile

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 检查文件大小
    contents = await file.read()
    if len(contents) > 10 * 1024 * 1024:  # 10 MB
        raise HTTPException(status_code=413, detail="File too large")
    
    # 检查文件类型
    if file.content_type not in ["image/jpeg", "image/png"]:
        raise HTTPException(status_code=415, detail="Unsupported file type")
    
    return {"filename": file.filename, "size": len(contents)}

测试

使用 TestClient

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_read_item():
    response = client.get("/items/foo")
    assert response.status_code == 200
    assert response.json() == {"name": "Foo"}

def test_read_item_not_found():
    response = client.get("/items/bar")
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}

def test_create_item_invalid():
    response = client.post("/items/", json={"name": "", "price": -10})
    assert response.status_code == 422

调试技巧

1. 启用调试模式

app = FastAPI(debug=True)

2. 查看请求详情

from fastapi import Request

@app.get("/debug")
async def debug(request: Request):
    return {
        "method": request.method,
        "url": str(request.url),
        "headers": dict(request.headers),
        "query_params": dict(request.query_params),
        "path_params": request.path_params,
        "client": request.client.host if request.client else None,
    }

3. 性能分析

import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

实验：完整错误处理

创建一个健壮的 API 端点：

from typing import Optional
import logging

logger = logging.getLogger(__name__)

@app.get("/safe-items/{item_id}")
async def safe_read_item(item_id: int, q: Optional[str] = None):
    try:
        logger.info(f"Fetching item {item_id}")
        
        # 参数验证
        if item_id < 0:
            raise HTTPException(status_code=400, detail="Invalid item ID")
        
        # 业务逻辑
        item = get_item_from_db(item_id)
        
        if not item:
            logger.warning(f"Item {item_id} not found")
            raise HTTPException(status_code=404, detail="Item not found")
        
        # 搜索过滤
        if q and q.lower() not in item.name.lower():
            return {"message": "No match"}
        
        logger.info(f"Successfully returned item {item_id}")
        return item
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

验收标准

✅ 404、422、500 等错误正确返回
✅ 日志记录关键操作和错误
✅ 边界情况（空值、负数、大文件）正确处理
✅ 测试覆盖正常和异常路径
✅ 调试信息清晰可追踪

下一步：查看练习答案验证理解。