Python FastAPI 8: Xử lý lỗi

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    task = _tasks_db.get(task_id)
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task không tồn tại",
            headers={"X-Error-Code": "TASK_NOT_FOUND"},
        )
    return task

{
  "detail": "Task không tồn tại"
}

{
  "type": "TASK_NOT_FOUND",
  "title": "Task không tồn tại",
  "status": 404,
  "detail": "Task với id=42 không tồn tại hoặc đã bị xoá",
  "instance": "/api/v1/tasks/42",
  "request_id": "7f3a8b2c-...",
  "errors": [
    {
      "field": "title",
      "message": "Tối thiểu 1 ký tự"
    }
  ]
}

# app/exceptions.py
from typing import Optional, Any

class AppException(Exception):
    """Base exception cho mọi lỗi của app"""
    status_code: int = 500
    error_code: str = "INTERNAL_ERROR"
    title: str = "Lỗi không xác định"

    def __init__(
        self,
        detail: Optional[str] = None,
        extra: Optional[dict] = None,
    ):
        self.detail = detail or self.title
        self.extra = extra or {}
        super().__init__(self.detail)


# ===== 4xx: lỗi client =====
class NotFoundError(AppException):
    status_code = 404
    error_code = "NOT_FOUND"
    title = "Không tìm thấy tài nguyên"

class TaskNotFoundError(NotFoundError):
    error_code = "TASK_NOT_FOUND"
    title = "Task không tồn tại"

    def __init__(self, task_id: int):
        super().__init__(
            detail=f"Task với id={task_id} không tồn tại",
            extra={"task_id": task_id},
        )

class UnauthorizedError(AppException):
    status_code = 401
    error_code = "UNAUTHORIZED"
    title = "Chưa xác thực"

class ForbiddenError(AppException):
    status_code = 403
    error_code = "FORBIDDEN"
    title = "Không có quyền truy cập"

class ConflictError(AppException):
    status_code = 409
    error_code = "CONFLICT"
    title = "Xung đột dữ liệu"

class DuplicateTaskTitleError(ConflictError):
    error_code = "DUPLICATE_TASK_TITLE"
    title = "Tiêu đề task đã tồn tại"

    def __init__(self, title: str):
        super().__init__(
            detail=f"Đã có task với tiêu đề '{title}'",
            extra={"duplicate_title": title},
        )


# ===== Business rule errors =====
class BusinessRuleError(AppException):
    status_code = 422
    error_code = "BUSINESS_RULE_VIOLATION"
    title = "Vi phạm ràng buộc nghiệp vụ"

class InvalidStatusTransitionError(BusinessRuleError):
    error_code = "INVALID_STATUS_TRANSITION"
    title = "Không thể chuyển trạng thái"

    def __init__(self, from_status: str, to_status: str):
        super().__init__(
            detail=f"Không thể chuyển task từ {from_status} sang {to_status}",
            extra={"from": from_status, "to": to_status},
        )


# ===== 5xx: lỗi server =====
class ServiceUnavailableError(AppException):
    status_code = 503
    error_code = "SERVICE_UNAVAILABLE"
    title = "Dịch vụ tạm thời không khả dụng"

class ExternalAPIError(AppException):
    status_code = 502
    error_code = "EXTERNAL_API_ERROR"
    title = "Lỗi từ dịch vụ bên ngoài"

# app/error_handlers.py
import logging
import traceback
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from app.exceptions import AppException

logger = logging.getLogger("taskapi")


def _error_response(
    request: Request,
    status_code: int,
    error_code: str,
    title: str,
    detail: str,
    errors: list = None,
    extra: dict = None,
) -> JSONResponse:
    """Helper tạo response lỗi theo format thống nhất"""
    body = {
        "type": error_code,
        "title": title,
        "status": status_code,
        "detail": detail,
        "instance": str(request.url.path),
        "request_id": request.headers.get("X-Request-ID", "unknown"),
    }
    if errors:
        body["errors"] = errors
    if extra:
        body["extra"] = extra

    return JSONResponse(status_code=status_code, content=body)


def register_exception_handlers(app: FastAPI):
    """Đăng ký toàn bộ exception handlers"""

    # Handler cho custom AppException (và mọi subclass)
    @app.exception_handler(AppException)
    async def app_exception_handler(request: Request, exc: AppException):
        # Log warning cho 4xx, error cho 5xx
        log_fn = logger.error if exc.status_code >= 500 else logger.warning
        log_fn(
            f"[{exc.error_code}] {exc.detail} "
            f"at {request.method} {request.url.path}",
            extra={"error_extra": exc.extra},
        )

        return _error_response(
            request=request,
            status_code=exc.status_code,
            error_code=exc.error_code,
            title=exc.title,
            detail=exc.detail,
            extra=exc.extra if exc.extra else None,
        )

    # Handler cho lỗi validation (Pydantic)
    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(
        request: Request,
        exc: RequestValidationError,
    ):
        errors = []
        for err in exc.errors():
            # loc là tuple, vd: ("body", "title") hoặc ("query", "page")
            location = ".".join(str(x) for x in err["loc"][1:])
            errors.append({
                "field": location,
                "message": err["msg"],
                "type": err["type"],
            })

        logger.info(
            f"Validation error at {request.url.path}: {len(errors)} fields"
        )

        return _error_response(
            request=request,
            status_code=422,
            error_code="VALIDATION_ERROR",
            title="Dữ liệu không hợp lệ",
            detail="Một hoặc nhiều field không đạt yêu cầu",
            errors=errors,
        )

    # Handler cho HTTPException (FastAPI built-in)
    @app.exception_handler(StarletteHTTPException)
    async def http_exception_handler(
        request: Request,
        exc: StarletteHTTPException,
    ):
        # Map status code → error_code
        error_code_map = {
            401: "UNAUTHORIZED",
            403: "FORBIDDEN",
            404: "NOT_FOUND",
            405: "METHOD_NOT_ALLOWED",
            429: "TOO_MANY_REQUESTS",
        }

        return _error_response(
            request=request,
            status_code=exc.status_code,
            error_code=error_code_map.get(exc.status_code, "HTTP_ERROR"),
            title=str(exc.detail),
            detail=str(exc.detail),
        )

    # Handler cho MỌI exception khác (chưa được handle)
    @app.exception_handler(Exception)
    async def unhandled_exception_handler(request: Request, exc: Exception):
        # Log đầy đủ traceback để debug
        logger.exception(
            f"Unhandled exception at {request.method} {request.url.path}: "
            f"{type(exc).__name__}: {exc}"
        )

        # KHÔNG BAO GIỜ leak chi tiết nội bộ ra client
        return _error_response(
            request=request,
            status_code=500,
            error_code="INTERNAL_SERVER_ERROR",
            title="Lỗi hệ thống",
            detail="Đã có lỗi xảy ra. Vui lòng liên hệ support với request_id.",
        )

# app/main.py
from fastapi import FastAPI
from app.error_handlers import register_exception_handlers
from app.routers import tasks

app = FastAPI(title="TaskAPI")
register_exception_handlers(app)
app.include_router(tasks.router, prefix="/api/v1")

# app/services/task_service.py
from app.exceptions import (
    TaskNotFoundError,
    DuplicateTaskTitleError,
    InvalidStatusTransitionError,
)

_VALID_TRANSITIONS = {
    "pending": ["in_progress", "done"],
    "in_progress": ["pending", "done"],
    "done": [],  # done là trạng thái cuối, không chuyển đi đâu nữa
}

class TaskService:
    def __init__(self, db):
        self.db = db

    def get_task(self, task_id: int) -> dict:
        task = self.db.get(task_id)
        if not task:
            raise TaskNotFoundError(task_id)
        return task

    def create_task(self, data: dict) -> dict:
        if self._title_exists(data["title"]):
            raise DuplicateTaskTitleError(data["title"])
        return self._save(data)

    def change_status(self, task_id: int, new_status: str) -> dict:
        task = self.get_task(task_id)
        current = task["status"]

        if new_status not in _VALID_TRANSITIONS.get(current, []):
            raise InvalidStatusTransitionError(current, new_status)

        task["status"] = new_status
        return task

# app/routers/tasks.py
from fastapi import APIRouter, Depends
from app.services.task_service import TaskService
from app.models.task import TaskCreate, TaskResponse

router = APIRouter(prefix="/tasks", tags=["tasks"])

@router.get("/{task_id}", response_model=TaskResponse)
def get_task(
    task_id: int,
    service: TaskService = Depends(get_task_service),
):
    # Nếu task không tồn tại, service raise TaskNotFoundError
    # Exception handler tự xử lý, trả về 404 đúng format
    return service.get_task(task_id)

@router.post("/", response_model=TaskResponse, status_code=201)
def create_task(
    task: TaskCreate,
    service: TaskService = Depends(get_task_service),
):
    return service.create_task(task.model_dump())

@router.patch("/{task_id}/status", response_model=TaskResponse)
def change_status(
    task_id: int,
    new_status: str,
    service: TaskService = Depends(get_task_service),
):
    return service.change_status(task_id, new_status)

{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "title"],
      "msg": "String should have at least 1 character",
      "input": "",
      "ctx": {"min_length": 1}
    }
  ]
}

{
  "type": "VALIDATION_ERROR",
  "title": "Dữ liệu không hợp lệ",
  "status": 422,
  "detail": "Một hoặc nhiều field không đạt yêu cầu",
  "instance": "/api/v1/tasks",
  "request_id": "7f3a8b2c-...",
  "errors": [
    {
      "field": "title",
      "message": "String should have at least 1 character",
      "type": "string_too_short"
    },
    {
      "field": "priority",
      "message": "Input should be 'low', 'medium' or 'high'",
      "type": "enum"
    }
  ]
}

# app/i18n.py
_ERROR_MESSAGES = {
    "string_too_short": "Độ dài tối thiểu là {min_length} ký tự",
    "string_too_long": "Độ dài tối đa là {max_length} ký tự",
    "greater_than": "Phải lớn hơn {gt}",
    "less_than_equal": "Phải nhỏ hơn hoặc bằng {le}",
    "missing": "Field bắt buộc",
    "int_parsing": "Phải là số nguyên",
    "enum": "Giá trị không hợp lệ, phải là một trong: {expected}",
}

def translate_validation_error(err: dict) -> str:
    err_type = err["type"]
    template = _ERROR_MESSAGES.get(err_type)
    if not template:
        return err["msg"]

    ctx = err.get("ctx", {})
    try:
        return template.format(**ctx)
    except (KeyError, IndexError):
        return err["msg"]

from app.i18n import translate_validation_error

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    errors = []
    for err in exc.errors():
        location = ".".join(str(x) for x in err["loc"][1:])
        errors.append({
            "field": location,
            "message": translate_validation_error(err),  # <-- tiếng Việt
            "type": err["type"],
        })
    ...

import httpx
from app.exceptions import ExternalAPIError, ServiceUnavailableError

async def send_notification(user_email: str, message: str):
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.post(
                "https://api.notifications.example.com/send",
                json={"to": user_email, "message": message},
            )
            response.raise_for_status()
            return response.json()

    except httpx.TimeoutException:
        # Log với đầy đủ context, raise lỗi friendly với client
        logger.error(f"Notification timeout for {user_email}")
        raise ServiceUnavailableError(
            detail="Dịch vụ notification đang phản hồi chậm",
        )

    except httpx.HTTPStatusError as e:
        logger.error(
            f"Notification API returned {e.response.status_code}: "
            f"{e.response.text}"
        )
        raise ExternalAPIError(
            detail="Dịch vụ notification trả lỗi",
            extra={"upstream_status": e.response.status_code},
        )

    except httpx.RequestError as e:
        logger.exception(f"Cannot reach notification API: {e}")
        raise ServiceUnavailableError(
            detail="Không thể kết nối dịch vụ notification",
        )

pip install tenacity

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
import httpx

@retry(
    stop=stop_after_attempt(3),                         # Tối đa 3 lần
    wait=wait_exponential(multiplier=1, min=1, max=10), # 1s, 2s, 4s, 8s...
    retry=retry_if_exception_type((
        httpx.TimeoutException,
        httpx.ConnectError,
    )),                                                 # Chỉ retry với lỗi network
)
async def fetch_user_data(user_id: int) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        response.raise_for_status()
        return response.json()

pip install purgatory

from purgatory import AsyncCircuitBreakerFactory

circuitbreaker = AsyncCircuitBreakerFactory(
    default_threshold=5,      # Sau 5 lỗi liên tiếp...
    default_ttl=30,           # ...ngắt mạch 30 giây
)

async def call_external_service(user_id: int):
    async with await circuitbreaker.get_breaker("notifications") as breaker:
        async with breaker.context():
            return await fetch_user_data(user_id)

# app/logging_config.py
import logging
import sys
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno,
        }

        # Thêm exception info nếu có
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        # Thêm extra data nếu logger được gọi với extra={...}
        for key, value in record.__dict__.items():
            if key not in {
                "name", "msg", "args", "created", "filename", "funcName",
                "levelname", "levelno", "lineno", "module", "msecs",
                "message", "pathname", "process", "processName",
                "relativeCreated", "thread", "threadName", "exc_info",
                "exc_text", "stack_info",
            }:
                log_data[key] = value

        return json.dumps(log_data, ensure_ascii=False)


def setup_logging(level: str = "INFO"):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JSONFormatter())

    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(level)

    # Giảm độ nhiều của thư viện bên thứ ba
    logging.getLogger("uvicorn.access").setLevel("WARNING")
    logging.getLogger("httpx").setLevel("WARNING")

from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError
from app.exceptions import (
    ConflictError,
    ServiceUnavailableError,
    AppException,
)

class DatabaseError(AppException):
    status_code = 500
    error_code = "DATABASE_ERROR"
    title = "Lỗi cơ sở dữ liệu"


@app.exception_handler(IntegrityError)
async def integrity_error_handler(request, exc):
    """Lỗi ràng buộc DB (unique, foreign key...)"""
    logger.warning(f"Integrity error: {exc}", extra={"query": str(exc.statement)})

    # Check loại lỗi từ message (PostgreSQL, MySQL có format khác nhau)
    msg = str(exc.orig).lower()

    if "unique" in msg or "duplicate" in msg:
        raise ConflictError(detail="Dữ liệu đã tồn tại")
    if "foreign key" in msg:
        raise ConflictError(detail="Vi phạm ràng buộc khoá ngoại")

    # Lỗi integrity khác
    raise DatabaseError(detail="Vi phạm ràng buộc cơ sở dữ liệu")


@app.exception_handler(OperationalError)
async def operational_error_handler(request, exc):
    """DB down, timeout, connection lost..."""
    logger.error(f"DB operational error: {exc}")
    raise ServiceUnavailableError(
        detail="Cơ sở dữ liệu tạm thời không khả dụng"
    )


@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_error_handler(request, exc):
    """Mọi lỗi SQLAlchemy khác"""
    logger.exception(f"Unhandled DB error: {exc}")
    raise DatabaseError(detail="Lỗi cơ sở dữ liệu")

# HTTP client
async with httpx.AsyncClient(timeout=5.0) as client:
    ...

# Database query (với SQLAlchemy)
engine = create_engine(
    DATABASE_URL,
    connect_args={"connect_timeout": 5},
    pool_pre_ping=True,
    pool_recycle=3600,
)

# Redis
redis_client = Redis(
    host="localhost",
    socket_timeout=3,
    socket_connect_timeout=3,
)

# Uvicorn timeout cho request xử lý quá lâu
# uvicorn app.main:app --timeout-keep-alive 30

pip install "sentry-sdk[fastapi]"

# app/main.py
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from app.config import settings

if settings.sentry_dsn:
    sentry_sdk.init(
        dsn=settings.sentry_dsn,
        environment=settings.environment,  # production, staging...
        traces_sample_rate=0.1,  # Sample 10% request để performance monitoring
        integrations=[FastApiIntegration()],
    )

# Và thế là xong - mọi unhandled exception sẽ tự động gửi lên Sentry

# tests/test_tasks.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_get_nonexistent_task_returns_404():
    response = client.get("/api/v1/tasks/99999")

    assert response.status_code == 404
    body = response.json()
    assert body["type"] == "TASK_NOT_FOUND"
    assert body["status"] == 404
    assert "99999" in body["detail"]

def test_create_task_with_invalid_data_returns_422():
    response = client.post(
        "/api/v1/tasks",
        json={"title": "", "priority": "super-high"},
    )

    assert response.status_code == 422
    body = response.json()
    assert body["type"] == "VALIDATION_ERROR"
    assert len(body["errors"]) >= 2

    fields = {e["field"] for e in body["errors"]}
    assert "title" in fields
    assert "priority" in fields

def test_duplicate_title_returns_409():
    client.post("/api/v1/tasks", json={"title": "Unique Task"})
    response = client.post("/api/v1/tasks", json={"title": "Unique Task"})

    assert response.status_code == 409
    assert response.json()["type"] == "DUPLICATE_TASK_TITLE"