Back to Details

error-handling

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Error Handling

错误处理

Production-ready error handling for Python APIs using the Let it crash philosophy.
基于「让它崩溃」理念的Python API生产级错误处理方案。

Design Philosophy

设计理念

Let it crash - Don't be defensive. Let exceptions propagate naturally and handle them at boundaries.
python
undefined
让它崩溃 - 不要过度防御。让异常自然传播，在边界处处理。
python
undefined

BAD - Too defensive, obscures errors

错误示例 - 过度防御，掩盖了错误

@app.get("/users/{user_id}") async def get_user(user_id: int): try: user = await user_service.get(user_id) if not user: raise HTTPException(404, "Not found") return user except DatabaseError as e: raise HTTPException(500, "Database error") except Exception as e: logger.exception("Unexpected error") raise HTTPException(500, "Internal error")
@app.get("/users/{user_id}") async def get_user(user_id: int): try: user = await user_service.get(user_id) if not user: raise HTTPException(404, "Not found") return user except DatabaseError as e: raise HTTPException(500, "Database error") except Exception as e: logger.exception("Unexpected error") raise HTTPException(500, "Internal error")

GOOD - Let exceptions propagate, handle at boundary

正确示例 - 让异常自然传播，在边界处处理

@app.get("/users/{user_id}") async def get_user(user_id: int): user = await user_service.get(user_id) if not user: raise UserNotFoundError(user_id) return user
undefined
@app.get("/users/{user_id}") async def get_user(user_id: int): user = await user_service.get(user_id) if not user: raise UserNotFoundError(user_id) return user
undefined

Core Principles

核心原则

  1. Raise low, catch high - Throw exceptions where errors occur, handle at API boundaries
  2. Domain exceptions - Create semantic exceptions, not generic ones
  3. Global handlers - Use 
    @app.exception_handler()
    for centralized error formatting
  4. No bare except - Always catch specific exceptions
  5. Preserve context - Use 
    raise ... from error
    to keep original traceback
  1. 低层抛出，高层捕获 - 在错误发生处抛出异常，在API边界处处理
  2. 领域异常 - 创建语义化异常，而非通用异常
  3. 全局处理器 - 使用
    @app.exception_handler()
    实现集中式错误格式化
  4. 不使用裸except - 始终捕获特定类型的异常
  5. 保留上下文 - 使用
    raise ... from error
    保留原始堆栈跟踪

Quick Start

快速开始

1. Define Domain Exceptions

1. 定义领域异常

python
from enum import StrEnum

class ErrorCode(StrEnum):
    USER_NOT_FOUND = "user_not_found"
    INVALID_CREDENTIALS = "invalid_credentials"
    RATE_LIMITED = "rate_limited"

class DomainError(Exception):
    """Base exception for all domain errors."""
    def __init__(self, code: ErrorCode, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code
        super().__init__(message)

class UserNotFoundError(DomainError):
    def __init__(self, user_id: int):
        super().__init__(
            code=ErrorCode.USER_NOT_FOUND,
            message=f"User {user_id} not found",
            status_code=404
        )
python
from enum import StrEnum

class ErrorCode(StrEnum):
    USER_NOT_FOUND = "user_not_found"
    INVALID_CREDENTIALS = "invalid_credentials"
    RATE_LIMITED = "rate_limited"

class DomainError(Exception):
    """所有领域异常的基类。"""
    def __init__(self, code: ErrorCode, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code
        super().__init__(message)

class UserNotFoundError(DomainError):
    def __init__(self, user_id: int):
        super().__init__(
            code=ErrorCode.USER_NOT_FOUND,
            message=f"User {user_id} not found",
            status_code=404
        )

2. Define Error Response Schema

2. 定义错误响应 Schema

python
from pydantic import BaseModel

class ErrorDetail(BaseModel):
    code: str
    message: str
    request_id: str | None = None

class ErrorResponse(BaseModel):
    error: ErrorDetail
python
from pydantic import BaseModel

class ErrorDetail(BaseModel):
    code: str
    message: str
    request_id: str | None = None

class ErrorResponse(BaseModel):
    error: ErrorDetail

3. Register Global Handlers

3. 注册全局处理器

python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": exc.code, "message": exc.message}}
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": "http_error", "message": str(exc.detail)}}
    )

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "message": "Invalid request"}}
    )

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
    # Log full error internally
    logger.exception("Unhandled error")
    # Return safe message to client
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "internal_error", "message": "Internal server error"}}
    )
python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": exc.code, "message": exc.message}}
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": "http_error", "message": str(exc.detail)}}
    )

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "message": "Invalid request"}}
    )

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
    # 内部记录完整错误
    logger.exception("Unhandled error")
    # 向客户端返回安全的错误信息
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "internal_error", "message": "Internal server error"}}
    )

4. Use in Routes

4. 在路由中使用

python
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user
python
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

When to Catch Exceptions

何时捕获异常

Only catch exceptions in these cases:
SituationExample
Need to retry
tenacity.retry()
for transient failures
Need to transformWrap third-party SDK errors as domain errors
Need to clean upUse 
finally
or context managers
Need to add context
raise DomainError(...) from original
仅在以下场景捕获异常：
场景示例
需要重试使用
tenacity.retry()
处理临时故障
需要转换异常将第三方SDK错误包装为领域异常
需要清理资源使用
finally
或上下文管理器
需要添加上下文信息使用
raise DomainError(...) from original

Python + FastAPI Integration

Python + FastAPI 集成

LayerResponsibility
Service/DomainRaise domain exceptions (
UserNotFoundError
)
RoutesLet exceptions propagate (no try/except)
Exception HandlersTransform to HTTP responses
MiddlewareAdd request context (request_id, timing)
层级职责
服务/领域层抛出领域异常（如
UserNotFoundError
）
路由层让异常自然传播（不使用try/except）
异常处理器将异常转换为HTTP响应
中间件添加请求上下文（request_id、计时等）

Common Patterns

常见模式

Third-Party SDK Wrapping

第三方SDK包装

python
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ExternalServiceError(DomainError):
    def __init__(self, service: str, original: Exception):
        super().__init__(
            code=ErrorCode.EXTERNAL_SERVICE_ERROR,
            message=f"{service} unavailable",
            status_code=503
        )
        self.__cause__ = original

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def call_payment_api(data: dict):
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post("https://api.payment.com/charge", json=data)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        raise ExternalServiceError("Payment API", e) from e
python
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ExternalServiceError(DomainError):
    def __init__(self, service: str, original: Exception):
        super().__init__(
            code=ErrorCode.EXTERNAL_SERVICE_ERROR,
            message=f"{service} unavailable",
            status_code=503
        )
        self.__cause__ = original

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def call_payment_api(data: dict):
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post("https://api.payment.com/charge", json=data)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        raise ExternalServiceError("Payment API", e) from e

Background Task Error Handling

后台任务错误处理

python
from fastapi import BackgroundTasks

async def safe_background_task(task_func, *args, **kwargs):
    try:
        await task_func(*args, **kwargs)
    except Exception as e:
        logger.exception(f"Background task failed: {e}")
        # Optional: send to dead letter queue or alerting

@app.post("/orders")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    result = await order_service.create(order)
    background_tasks.add_task(safe_background_task, send_confirmation_email, result.id)
    return result
python
from fastapi import BackgroundTasks

async def safe_background_task(task_func, *args, **kwargs):
    try:
        await task_func(*args, **kwargs)
    except Exception as e:
        logger.exception(f"Background task failed: {e}")
        # 可选：发送到死信队列或触发告警

@app.post("/orders")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    result = await order_service.create(order)
    background_tasks.add_task(safe_background_task, send_confirmation_email, result.id)
    return result

Troubleshooting

问题排查

IssueCauseFix
Stack trace in responseNo generic handlerAdd 
@app.exception_handler(Exception)
Lost original errorMissing 
from
Use 
raise NewError() from original
Validation errors leakDefault handlerOverride 
RequestValidationError
handler
Silent failuresSwallowed exceptionsLet exceptions propagate, handle at boundary
问题原因解决方法
响应中包含堆栈跟踪没有通用异常处理器添加
@app.exception_handler(Exception)
原始错误丢失未使用
from
关键字		使用
raise NewError() from original
验证错误泄露使用了默认处理器重写
RequestValidationError
处理器
静默失败异常被吞掉让异常自然传播，在边界处处理

References

参考资料

  • Python Patterns - Exception design, when to catch, SDK wrapping
  • FastAPI Patterns - HTTPException, global handlers, middleware
  • Pydantic Patterns - ValidationError, raise in validators
  • Asyncio Patterns - TaskGroup, timeout, background tasks
  • FastAPI Docs: Handling Errors
  • Pydantic Docs: Error Handling
  • Python Patterns - 异常设计、捕获时机、SDK包装
  • FastAPI Patterns - HTTPException、全局处理器、中间件
  • Pydantic Patterns - ValidationError、验证器中抛出异常
  • Asyncio Patterns - TaskGroup、超时、后台任务
  • FastAPI Docs: Handling Errors
  • Pydantic Docs: Error Handling