错误处理Skill error-handling

错误处理技能

专家级的结构化错误处理，适用于FastAPI后端和React/Next.js前端，提供一致的错误消息和日志记录。

快速参考

模式	后端	前端
自定义异常	class FeeNotPaidError(AppException)	N/A
尝试-捕捉	try: ... except SpecificError:	try { } catch (e) { }
全局处理器	@app.exception_handler	ErrorBoundary组件
用户消息	响应中的detail字段	Toast/Snackbar
错误日志	logger.error(...)	Console/Sentry

自定义异常（后端）

基础异常层级

# backend/app/errors/exceptions.py
from fastapi import HTTPException
from typing import Any

class AppException(HTTPException):
    """基础应用程序异常，带有用户友好的消息。"""

    def __init__(
        self,
        status_code: int,
        detail: str,
        user_message: str,
        headers: dict[str, Any] | None = None,
    ):
        self.user_message = user_message
        super().__init__(status_code=status_code, detail=detail, headers=headers)

class NotFoundError(AppException):
    """资源未找到（404）。"""

    def __init__(self, resource: str, identifier: str):
        super().__init__(
            status_code=404,
            detail=f"{resource} with id '{identifier}' not found",
            user_message=f"{resource} not found. Please check and try again.",
        )

class ValidationError(AppException):
    """验证失败（400）。"""

    def __init__(self, field: str, reason: str):
        super().__init__(
            status_code=400,
            detail=f"Validation error for field '{field}': {reason}",
            user_message=f"Invalid value for {field}. {reason}",
        )

class UnauthorizedError(AppException):
    """认证需要（401）。"""

    def __init__(self, reason: str = "Authentication required"):
        super().__init__(
            status_code=401,
            detail=reason,
            user_message="Please log in to continue.",
            headers={"WWW-Authenticate": "Bearer"},
        )

class ForbiddenError(AppException):
    """权限被拒绝（403）。"""

    def __init__(self, action: str):
        super().__init__(
            status_code=403,
            detail=f"Permission denied for action: {action}",
            user_message=f"You don't have permission to {action}.",
        )

class ConflictError(AppException):
    """资源冲突（409）。"""

    def __init__(self, resource: str, reason: str):
        super().__init__(
            status_code=409,
            detail=f"Conflict for {resource}: {reason}",
            user_message=f"{resource} conflict. {reason}",
        )

class RateLimitError(AppException):
    """请求过多（429）。"""

    def __init__(self, retry_after: int = 60):
        super().__init__(
            status_code=429,
            detail=f"Rate limit exceeded. Retry after {retry_after} seconds.",
            user_message="Too many requests. Please wait a moment and try again.",
            headers={"Retry-After": str(retry_after)},
        )

特定领域的异常

# backend/app/errors/domains.py
from .exceptions import NotFoundError, ValidationError, ConflictError

class StudentNotFoundError(NotFoundError):
    def __init__(self, student_id: int):
        super().__init__(resource="Student", identifier=str(student_id))

class FeeNotPaidError(ConflictError):
    def __init__(self, student_id: int, amount_due: float):
        super().__init__(
            resource="Fee",
            reason=f"Student {student_id} has unpaid fee of ${amount_due:.2f}",
        )

class AttendanceAlreadyMarkedError(ConflictError):
    def __init__(self, student_id: int, date: str):
        super().__init__(
            resource="Attendance",
            reason=f"Attendance already marked for student {student_id} on {date}",
        )

class InvalidGradeError(ValidationError):
    def __init__(self, grade: str, valid_grades: list[str]):
        super().__init__(
            field="grade",
            reason=f"'{grade}' is not valid. Must be one of: {', '.join(valid_grades)}",
        )

class InsufficientBalanceError(ValidationError):
    def __init__(self, required: float, available: float):
        super().__init__(
            field="amount",
            reason=f"Insufficient balance. Required: ${required:.2f}, Available: ${available:.2f}",
        )

尝试-捕捉模式

狭窄的尝试块

# 好的：特定异常，狭窄的范围
try:
    student = await get_student_by_id(student_id)
except StudentNotFoundError:
    raise StudentNotFoundError(student_id)

# 坏的：太宽泛，捕获一切
try:
    student = await get_student_by_id(student_id)
    calculate_fees(student)
    send_notification(student)
    update_records(student)
except Exception:
    pass  # 吞没！

清理与最后

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def database_connection():
    conn = await get_db_connection()
    try:
        yield conn
    except Exception as e:
        await conn.rollback()
        raise
    finally:
        await conn.close()

async def transfer_funds(from_account: int, to_account: int, amount: float):
    async with database_connection() as conn:
        try:
            await conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, from_account)
            await conn.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, to_account)
            await conn.commit()
        except InsufficientBalanceError:
            await conn.rollback()
            raise

错误日志（后端）

结构化JSON日志

# backend/app/core/logger.py
import logging
import json
from datetime import datetime
from typing import Any
from contextvars import ContextVar
import traceback

# 相关ID请求跟踪
correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")

class StructuredLogger:
    """结构化的JSON日志记录器，带有相关ID。"""

    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)

    def _log_record(self, level: str, message: str, extra: dict[str, Any] = None):
        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": level,
            "message": message,
            "correlation_id": correlation_id_var.get(),
            **(extra or {}),
        }
        return json.dumps(record)

    def info(self, message: str, **extra):
        self.logger.info(self._log_record("INFO", message, extra))

    def warning(self, message: str, **extra):
        self.logger.warning(self._log_record("WARNING", message, extra))

    def error(self, message: str, error: Exception = None, **extra):
        log_extra = {**extra}
        if error:
            log_extra["error_type"] = type(error).__name__
            log_extra["error_message"] = str(error)
            log_extra["stack_trace"] = traceback.format_exc()
        self.logger.error(self._log_record("ERROR", message, log_extra))

logger = StructuredLogger(__name__)

使用日志记录器

from app.core.logger import logger, correlation_id_var

async def get_student(student_id: int) -> Student:
    logger.info("Fetching student", student_id=student_id)

    try:
        student = await db.get_student(student_id)
        logger.info("Student fetched successfully", student_id=student_id, has_fees=bool(student.fees))
        return student
    except StudentNotFoundError:
        logger.warning("Student not found", student_id=student_id)
        raise
    except Exception as e:
        logger.error("Unexpected error fetching student", error=e, student_id=student_id)
        raise  # 重新抛出以处理

全局错误处理器（后端）

# backend/app/middleware/error_handler.py
from fastapi import Request, FastAPI
from fastapi.responses import JSONResponse
from app.errors.exceptions import AppException
from app.core.logger import logger

def setup_error_handlers(app: FastAPI):
    @app.exception_handler(AppException)
    async def app_exception_handler(request: Request, exc: AppException):
        logger.error(
            "App exception occurred",
            error=exc,
            status_code=exc.status_code,
            path=request.url.path,
        )
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "error": {
                    "code": exc.__class__.__name__,
                    "message": exc.user_message,
                    "internal": exc.detail,  # 只有在调试模式下
                }
            },
            headers=exc.headers,
        )

    @app.exception_handler(Exception)
    async def general_exception_handler(request: Request, exc: Exception):
        logger.error(
            "Unhandled exception",
            error=exc,
            path=request.url.path,
            method=request.method,
        )
        return JSONResponse(
            status_code=500,
            content={
                "error": {
                    "code": "INTERNAL_ERROR",
                    "message": "Something went wrong. Please try again later.",
                }
            },
        )

前端错误处理

错误类型

// frontend/lib/errors.ts
export interface ApiError {
  error: {
    code: string;
    message: string;
    internal?: string;
  };
}

export class ApiResponseError extends Error {
  code: string;
  userMessage: string;
  internalMessage?: string;
  statusCode: number;

  constructor(error: ApiError["error"], statusCode: number) {
    super(error.message);
    this.name = "ApiResponseError";
    this.code = error.code;
    this.userMessage = error.message;
    this.internalMessage = error.internal;
    this.statusCode = statusCode;
  }
}

错误映射器

// frontend/lib/errorMapper.ts
import { ApiResponseError } from "./errors";

export function getUserMessage(error: unknown): string {
  if (error instanceof ApiResponseError) {
    return error.userMessage;
  }

  if (error instanceof Error) {
    // 网络错误
    if (error.name === "TypeError" && error.message.includes("fetch")) {
      return "Unable to connect to server. Please check your internet connection.";
    }

    // 意外错误
    return "An unexpected error occurred. Please try again.";
  }

  return "An unknown error occurred.";
}

export function getErrorTitle(error: unknown): string {
  if (error instanceof ApiResponseError) {
    switch (error.statusCode) {
      case 401:
        return "Authentication Required";
      case 403:
        return "Access Denied";
      case 404:
        return "Not Found";
      case 409:
        return "Conflict";
      case 422:
        return "Validation Error";
      case 429:
        return "Rate Limited";
      case 500:
        return "Server Error";
      default:
        return "Error";
    }
  }

  return "Error";
}

API客户端错误处理

// frontend/lib/api.ts
import { ApiResponseError } from "./errors";
import { getUserMessage } from "./errorMapper";

interface RequestOptions extends RequestInit {
  params?: Record<string, string>;
}

class ApiClient {
  private baseUrl: string;

  constructor(baseUrl: string = "/api/v1") {
    this.baseUrl = baseUrl;
  }

  async request<T>(endpoint: string, options: RequestOptions = {}): Promise<T> {
    const url = new URL(`${this.baseUrl}${endpoint}`, window.location.origin);

    if (options.params) {
      Object.entries(options.params).forEach(([key, value]) => {
        url.searchParams.append(key, value);
      });
    }

    const response = await fetch(url.toString(), {
      ...options,
      headers: {
        "Content-Type": "application/json",
        ...options.headers,
      },
    });

    if (!response.ok) {
      const errorData = await response.json().catch(() => ({}));

      throw new ApiResponseError(
        {
          code: errorData.error?.code || "HTTP_ERROR",
          message: errorData.error?.message || response.statusText,
          internal: errorData.error?.internal,
        },
        response.status
      );
    }

    if (response.status === 204) {
      return undefined as T;
    }

    return response.json() as Promise<T>;
  }

  get<T>(endpoint: string, params?: Record<string, string>): Promise<T> {
    return this.request<T>(endpoint, { method: "GET", params });
  }

  post<T>(endpoint: string, body: unknown): Promise<T> {
    return this.request<T>(endpoint, { method: "POST", body: JSON.stringify(body) });
  }

  put<T>(endpoint: string, body: unknown): Promise<T> {
    return this.request<T>(endpoint, { method: "PUT", body: JSON.stringify(body) });
  }

  delete<T>(endpoint: string): Promise<T> {
    return this.request<T>(endpoint, { method: "DELETE" });
  }
}

export const api = new ApiClient();

React Query错误处理

// frontend/hooks/useApi.ts
import { useQuery, useMutation, UseQueryOptions } from "@tanstack/react-query";
import { api } from "../lib/api";
import { ApiResponseError } from "../lib/errors";
import { toast } from "sonner";

export function useFetch<T>(
  key: string[],
  endpoint: string,
  options?: Partial<UseQueryOptions<T>>
) {
  return useQuery({
    queryKey: key,
    queryFn: () => api.get<T>(endpoint),
    ...options,
    onError: (error: unknown) => {
      if (error instanceof ApiResponseError) {
        toast.error(error.userMessage);
      } else {
        toast.error("Failed to fetch data");
      }
    },
  });
}

export function useMutationWithError<T, V>(
  mutationFn: (variables: V) => Promise<T>,
  successMessage: string,
  onSuccess?: (data: T) => void
) {
  return useMutation({
    mutationFn,
    onSuccess: (data) => {
      toast.success(successMessage);
      onSuccess?.(data);
    },
    onError: (error: unknown) => {
      if (error instanceof ApiResponseError) {
        toast.error(error.userMessage);
      } else {
        toast.error("An error occurred. Please try again.");
      }
    },
  });
}

质量检查表

• [ ] 没有吞没的错误：每个异常都被记录或呈现给用户
• [ ] 内部与用户消息分离：技术细节在detail中，用户友好的在user_message中
• [ ] 4xx与5xx正确映射：客户端错误（4xx）与服务器错误（5xx）
• [ ] 日志中不包含PII：永远不要记录密码、令牌、个人数据
• [ ] 相关ID：使用请求ID跨服务跟踪错误
• [ ] 一致的错误格式：所有API错误都具有相同的结构

错误响应格式

{
  "error": {
    "code": "STUDENT_NOT_FOUND",
    "message": "Student not found. Please check and try again.",
    "internal": "Student with id '12345' not found"
  }
}

集成点

技能	集成
@fastapi-app	main.py中的全局异常处理器
@api-route-design	响应中适当的状态代码
@jwt-auth	认证失败的UnauthorizedError
@db-migration	优雅地处理迁移错误
@env-config	配置验证错误