error-handling-patterns

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Error Handling Patterns

错误处理模式

Build resilient applications with robust error handling strategies that gracefully handle failures and provide excellent debugging experiences.
借助健壮的错误处理策略构建高韧性应用,优雅处理故障并提供出色的调试体验。

When to Use This Skill

何时使用此技能

  • Implementing error handling in new features
  • Designing error-resilient APIs
  • Debugging production issues
  • Improving application reliability
  • Creating better error messages for users and developers
  • Implementing retry and circuit breaker patterns
  • Handling async/concurrent errors
  • Building fault-tolerant distributed systems
  • 在新功能中实现错误处理
  • 设计具备错误韧性的API
  • 调试生产环境问题
  • 提升应用可靠性
  • 为用户和开发者创建更友好的错误提示
  • 实现重试和断路器模式
  • 处理异步/并发错误
  • 构建容错分布式系统

Core Concepts

核心概念

1. Error Handling Philosophies

1. 错误处理理念

Exceptions vs Result Types:
  • Exceptions: Traditional try-catch, disrupts control flow
  • Result Types: Explicit success/failure, functional approach
  • Error Codes: C-style, requires discipline
  • Option/Maybe Types: For nullable values
When to Use Each:
  • Exceptions: Unexpected errors, exceptional conditions
  • Result Types: Expected errors, validation failures
  • Panics/Crashes: Unrecoverable errors, programming bugs
异常 vs Result类型:
  • 异常(Exceptions):传统的try-catch,会中断控制流
  • Result类型(Result Types):显式的成功/失败,函数式编程方式
  • 错误码(Error Codes):C语言风格,需要严格规范
  • Option/Maybe类型:用于可空值
各方式适用场景:
  • 异常:意外错误、异常情况
  • Result类型:预期内的错误、验证失败
  • 恐慌/崩溃(Panics/Crashes):不可恢复的错误、编程漏洞

2. Error Categories

2. 错误分类

Recoverable Errors:
  • Network timeouts
  • Missing files
  • Invalid user input
  • API rate limits
Unrecoverable Errors:
  • Out of memory
  • Stack overflow
  • Programming bugs (null pointer, etc.)
可恢复错误:
  • 网络超时
  • 文件缺失
  • 无效用户输入
  • API速率限制
不可恢复错误:
  • 内存不足
  • 栈溢出
  • 编程漏洞(如空指针等)

Language-Specific Patterns

各语言特定模式

Python Error Handling

Python错误处理

Custom Exception Hierarchy:
python
class ApplicationError(Exception):
    """Base exception for all application errors."""
    def __init__(self, message: str, code: str = None, details: dict = None):
        super().__init__(message)
        self.code = code
        self.details = details or {}
        self.timestamp = datetime.utcnow()

class ValidationError(ApplicationError):
    """Raised when validation fails."""
    pass

class NotFoundError(ApplicationError):
    """Raised when resource not found."""
    pass

class ExternalServiceError(ApplicationError):
    """Raised when external service fails."""
    def __init__(self, message: str, service: str, **kwargs):
        super().__init__(message, **kwargs)
        self.service = service
自定义异常层级:
python
class ApplicationError(Exception):
    """Base exception for all application errors."""
    def __init__(self, message: str, code: str = None, details: dict = None):
        super().__init__(message)
        self.code = code
        self.details = details or {}
        self.timestamp = datetime.utcnow()

class ValidationError(ApplicationError):
    """Raised when validation fails."""
    pass

class NotFoundError(ApplicationError):
    """Raised when resource not found."""
    pass

class ExternalServiceError(ApplicationError):
    """Raised when external service fails."""
    def __init__(self, message: str, service: str, **kwargs):
        super().__init__(message, **kwargs)
        self.service = service

Usage

Usage

def get_user(user_id: str) -> User: user = db.query(User).filter_by(id=user_id).first() if not user: raise NotFoundError( f"User not found", code="USER_NOT_FOUND", details={"user_id": user_id} ) return user

**Context Managers for Cleanup:**

```python
from contextlib import contextmanager

@contextmanager
def database_transaction(session):
    """Ensure transaction is committed or rolled back."""
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()
def get_user(user_id: str) -> User: user = db.query(User).filter_by(id=user_id).first() if not user: raise NotFoundError( f"User not found", code="USER_NOT_FOUND", details={"user_id": user_id} ) return user

**用于清理的上下文管理器:**

```python
from contextlib import contextmanager

@contextmanager
def database_transaction(session):
    """Ensure transaction is committed or rolled back."""
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()

Usage

Usage

with database_transaction(db.session) as session: user = User(name="Alice") session.add(user) # Automatic commit or rollback

**Retry with Exponential Backoff:**

```python
import time
from functools import wraps
from typing import TypeVar, Callable

T = TypeVar('T')

def retry(
    max_attempts: int = 3,
    backoff_factor: float = 2.0,
    exceptions: tuple = (Exception,)
):
    """Retry decorator with exponential backoff."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        sleep_time = backoff_factor ** attempt
                        time.sleep(sleep_time)
                        continue
                    raise
            raise last_exception
        return wrapper
    return decorator
with database_transaction(db.session) as session: user = User(name="Alice") session.add(user) # Automatic commit or rollback

**指数退避重试:**

```python
import time
from functools import wraps
from typing import TypeVar, Callable

T = TypeVar('T')

def retry(
    max_attempts: int = 3,
    backoff_factor: float = 2.0,
    exceptions: tuple = (Exception,)
):
    """Retry decorator with exponential backoff."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        sleep_time = backoff_factor ** attempt
                        time.sleep(sleep_time)
                        continue
                    raise
            raise last_exception
        return wrapper
    return decorator

Usage

Usage

@retry(max_attempts=3, exceptions=(NetworkError,)) def fetch_data(url: str) -> dict: response = requests.get(url, timeout=5) response.raise_for_status() return response.json()
undefined
@retry(max_attempts=3, exceptions=(NetworkError,)) def fetch_data(url: str) -> dict: response = requests.get(url, timeout=5) response.raise_for_status() return response.json()
undefined

TypeScript/JavaScript Error Handling

TypeScript/JavaScript错误处理

Custom Error Classes:
typescript
// Custom error classes
class ApplicationError extends Error {
  constructor(
    message: string,
    public code: string,
    public statusCode: number = 500,
    public details?: Record<string, any>,
  ) {
    super(message);
    this.name = this.constructor.name;
    Error.captureStackTrace(this, this.constructor);
  }
}

class ValidationError extends ApplicationError {
  constructor(message: string, details?: Record<string, any>) {
    super(message, "VALIDATION_ERROR", 400, details);
  }
}

class NotFoundError extends ApplicationError {
  constructor(resource: string, id: string) {
    super(`${resource} not found`, "NOT_FOUND", 404, { resource, id });
  }
}

// Usage
function getUser(id: string): User {
  const user = users.find((u) => u.id === id);
  if (!user) {
    throw new NotFoundError("User", id);
  }
  return user;
}
Result Type Pattern:
typescript
// Result type for explicit error handling
type Result<T, E = Error> = { ok: true; value: T } | { ok: false; error: E };

// Helper functions
function Ok<T>(value: T): Result<T, never> {
  return { ok: true, value };
}

function Err<E>(error: E): Result<never, E> {
  return { ok: false, error };
}

// Usage
function parseJSON<T>(json: string): Result<T, SyntaxError> {
  try {
    const value = JSON.parse(json) as T;
    return Ok(value);
  } catch (error) {
    return Err(error as SyntaxError);
  }
}

// Consuming Result
const result = parseJSON<User>(userJson);
if (result.ok) {
  console.log(result.value.name);
} else {
  console.error("Parse failed:", result.error.message);
}

// Chaining Results
function chain<T, U, E>(
  result: Result<T, E>,
  fn: (value: T) => Result<U, E>,
): Result<U, E> {
  return result.ok ? fn(result.value) : result;
}
Async Error Handling:
typescript
// Async/await with proper error handling
async function fetchUserOrders(userId: string): Promise<Order[]> {
  try {
    const user = await getUser(userId);
    const orders = await getOrders(user.id);
    return orders;
  } catch (error) {
    if (error instanceof NotFoundError) {
      return []; // Return empty array for not found
    }
    if (error instanceof NetworkError) {
      // Retry logic
      return retryFetchOrders(userId);
    }
    // Re-throw unexpected errors
    throw error;
  }
}

// Promise error handling
function fetchData(url: string): Promise<Data> {
  return fetch(url)
    .then((response) => {
      if (!response.ok) {
        throw new NetworkError(`HTTP ${response.status}`);
      }
      return response.json();
    })
    .catch((error) => {
      console.error("Fetch failed:", error);
      throw error;
    });
}
自定义错误类:
typescript
// Custom error classes
class ApplicationError extends Error {
  constructor(
    message: string,
    public code: string,
    public statusCode: number = 500,
    public details?: Record<string, any>,
  ) {
    super(message);
    this.name = this.constructor.name;
    Error.captureStackTrace(this, this.constructor);
  }
}

class ValidationError extends ApplicationError {
  constructor(message: string, details?: Record<string, any>) {
    super(message, "VALIDATION_ERROR", 400, details);
  }
}

class NotFoundError extends ApplicationError {
  constructor(resource: string, id: string) {
    super(`${resource} not found`, "NOT_FOUND", 404, { resource, id });
  }
}

// Usage
function getUser(id: string): User {
  const user = users.find((u) => u.id === id);
  if (!user) {
    throw new NotFoundError("User", id);
  }
  return user;
}
Result类型模式:
typescript
// Result type for explicit error handling
type Result<T, E = Error> = { ok: true; value: T } | { ok: false; error: E };

// Helper functions
function Ok<T>(value: T): Result<T, never> {
  return { ok: true, value };
}

function Err<E>(error: E): Result<never, E> {
  return { ok: false, error };
}

// Usage
function parseJSON<T>(json: string): Result<T, SyntaxError> {
  try {
    const value = JSON.parse(json) as T;
    return Ok(value);
  } catch (error) {
    return Err(error as SyntaxError);
  }
}

// Consuming Result
const result = parseJSON<User>(userJson);
if (result.ok) {
  console.log(result.value.name);
} else {
  console.error("Parse failed:", result.error.message);
}

// Chaining Results
function chain<T, U, E>(
  result: Result<T, E>,
  fn: (value: T) => Result<U, E>,
): Result<U, E> {
  return result.ok ? fn(result.value) : result;
}
异步错误处理:
typescript
// Async/await with proper error handling
async function fetchUserOrders(userId: string): Promise<Order[]> {
  try {
    const user = await getUser(userId);
    const orders = await getOrders(user.id);
    return orders;
  } catch (error) {
    if (error instanceof NotFoundError) {
      return []; // Return empty array for not found
    }
    if (error instanceof NetworkError) {
      // Retry logic
      return retryFetchOrders(userId);
    }
    // Re-throw unexpected errors
    throw error;
  }
}

// Promise error handling
function fetchData(url: string): Promise<Data> {
  return fetch(url)
    .then((response) => {
      if (!response.ok) {
        throw new NetworkError(`HTTP ${response.status}`);
      }
      return response.json();
    })
    .catch((error) => {
      console.error("Fetch failed:", error);
      throw error;
    });
}

Rust Error Handling

Rust错误处理

Result and Option Types:
rust
use std::fs::File;
use std::io::{self, Read};

// Result type for operations that can fail
fn read_file(path: &str) -> Result<String, io::Error> {
    let mut file = File::open(path)?;  // ? operator propagates errors
    let mut contents = String::new();
    file.read_to_string(&mut contents)?;
    Ok(contents)
}

// Custom error types
#[derive(Debug)]
enum AppError {
    Io(io::Error),
    Parse(std::num::ParseIntError),
    NotFound(String),
    Validation(String),
}

impl From<io::Error> for AppError {
    fn from(error: io::Error) -> Self {
        AppError::Io(error)
    }
}

// Using custom error type
fn read_number_from_file(path: &str) -> Result<i32, AppError> {
    let contents = read_file(path)?;  // Auto-converts io::Error
    let number = contents.trim().parse()
        .map_err(AppError::Parse)?;   // Explicitly convert ParseIntError
    Ok(number)
}

// Option for nullable values
fn find_user(id: &str) -> Option<User> {
    users.iter().find(|u| u.id == id).cloned()
}

// Combining Option and Result
fn get_user_age(id: &str) -> Result<u32, AppError> {
    find_user(id)
        .ok_or_else(|| AppError::NotFound(id.to_string()))
        .map(|user| user.age)
}
Result和Option类型:
rust
use std::fs::File;
use std::io::{self, Read};

// Result type for operations that can fail
fn read_file(path: &str) -> Result<String, io::Error> {
    let mut file = File::open(path)?;  // ? operator propagates errors
    let mut contents = String::new();
    file.read_to_string(&mut contents)?;
    Ok(contents)
}

// Custom error types
#[derive(Debug)]
enum AppError {
    Io(io::Error),
    Parse(std::num::ParseIntError),
    NotFound(String),
    Validation(String),
}

impl From<io::Error> for AppError {
    fn from(error: io::Error) -> Self {
        AppError::Io(error)
    }
}

// Using custom error type
fn read_number_from_file(path: &str) -> Result<i32, AppError> {
    let contents = read_file(path)?;  // Auto-converts io::Error
    let number = contents.trim().parse()
        .map_err(AppError::Parse)?;   // Explicitly convert ParseIntError
    Ok(number)
}

// Option for nullable values
fn find_user(id: &str) -> Option<User> {
    users.iter().find(|u| u.id == id).cloned()
}

// Combining Option and Result
fn get_user_age(id: &str) -> Result<u32, AppError> {
    find_user(id)
        .ok_or_else(|| AppError::NotFound(id.to_string()))
        .map(|user| user.age)
}

Go Error Handling

Go错误处理

Explicit Error Returns:
go
// Basic error handling
func getUser(id string) (*User, error) {
    user, err := db.QueryUser(id)
    if err != nil {
        return nil, fmt.Errorf("failed to query user: %w", err)
    }
    if user == nil {
        return nil, errors.New("user not found")
    }
    return user, nil
}

// Custom error types
type ValidationError struct {
    Field   string
    Message string
}

func (e *ValidationError) Error() string {
    return fmt.Sprintf("validation failed for %s: %s", e.Field, e.Message)
}

// Sentinel errors for comparison
var (
    ErrNotFound     = errors.New("not found")
    ErrUnauthorized = errors.New("unauthorized")
    ErrInvalidInput = errors.New("invalid input")
)

// Error checking
user, err := getUser("123")
if err != nil {
    if errors.Is(err, ErrNotFound) {
        // Handle not found
    } else {
        // Handle other errors
    }
}

// Error wrapping and unwrapping
func processUser(id string) error {
    user, err := getUser(id)
    if err != nil {
        return fmt.Errorf("process user failed: %w", err)
    }
    // Process user
    return nil
}

// Unwrap errors
err := processUser("123")
if err != nil {
    var valErr *ValidationError
    if errors.As(err, &valErr) {
        fmt.Printf("Validation error: %s\n", valErr.Field)
    }
}
显式错误返回:
go
// Basic error handling
func getUser(id string) (*User, error) {
    user, err := db.QueryUser(id)
    if err != nil {
        return nil, fmt.Errorf("failed to query user: %w", err)
    }
    if user == nil {
        return nil, errors.New("user not found")
    }
    return user, nil
}

// Custom error types
type ValidationError struct {
    Field   string
    Message string
}

func (e *ValidationError) Error() string {
    return fmt.Sprintf("validation failed for %s: %s", e.Field, e.Message)
}

// Sentinel errors for comparison
var (
    ErrNotFound     = errors.New("not found")
    ErrUnauthorized = errors.New("unauthorized")
    ErrInvalidInput = errors.New("invalid input")
)

// Error checking
user, err := getUser("123")
if err != nil {
    if errors.Is(err, ErrNotFound) {
        // Handle not found
    } else {
        // Handle other errors
    }
}

// Error wrapping and unwrapping
func processUser(id string) error {
    user, err := getUser(id)
    if err != nil {
        return fmt.Errorf("process user failed: %w", err)
    }
    // Process user
    return nil
}

// Unwrap errors
err := processUser("123")
if err != nil {
    var valErr *ValidationError
    if errors.As(err, &valErr) {
        fmt.Printf("Validation error: %s\n", valErr.Field)
    }
}

Universal Patterns

通用模式

Pattern 1: Circuit Breaker

模式1:断路器

Prevent cascading failures in distributed systems.
python
from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, TypeVar

T = TypeVar('T')

class CircuitState(Enum):
    CLOSED = "closed"       # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        timeout: timedelta = timedelta(seconds=60),
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.success_threshold = success_threshold
        self.failure_count = 0
        self.success_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None

    def call(self, func: Callable[[], T]) -> T:
        if self.state == CircuitState.OPEN:
            if datetime.now() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
防止分布式系统中的级联故障。
python
from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, TypeVar

T = TypeVar('T')

class CircuitState(Enum):
    CLOSED = "closed"       # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        timeout: timedelta = timedelta(seconds=60),
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.success_threshold = success_threshold
        self.failure_count = 0
        self.success_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None

    def call(self, func: Callable[[], T]) -> T:
        if self.state == CircuitState.OPEN:
            if datetime.now() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.success_count = 0

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

Usage

Usage

circuit_breaker = CircuitBreaker()
def fetch_data(): return circuit_breaker.call(lambda: external_api.get_data())
undefined
circuit_breaker = CircuitBreaker()
def fetch_data(): return circuit_breaker.call(lambda: external_api.get_data())
undefined

Pattern 2: Error Aggregation

模式2:错误聚合

Collect multiple errors instead of failing on first error.
typescript
class ErrorCollector {
  private errors: Error[] = [];

  add(error: Error): void {
    this.errors.push(error);
  }

  hasErrors(): boolean {
    return this.errors.length > 0;
  }

  getErrors(): Error[] {
    return [...this.errors];
  }

  throw(): never {
    if (this.errors.length === 1) {
      throw this.errors[0];
    }
    throw new AggregateError(
      this.errors,
      `${this.errors.length} errors occurred`,
    );
  }
}

// Usage: Validate multiple fields
function validateUser(data: any): User {
  const errors = new ErrorCollector();

  if (!data.email) {
    errors.add(new ValidationError("Email is required"));
  } else if (!isValidEmail(data.email)) {
    errors.add(new ValidationError("Email is invalid"));
  }

  if (!data.name || data.name.length < 2) {
    errors.add(new ValidationError("Name must be at least 2 characters"));
  }

  if (!data.age || data.age < 18) {
    errors.add(new ValidationError("Age must be 18 or older"));
  }

  if (errors.hasErrors()) {
    errors.throw();
  }

  return data as User;
}
收集多个错误而非在第一个错误时就失败。
typescript
class ErrorCollector {
  private errors: Error[] = [];

  add(error: Error): void {
    this.errors.push(error);
  }

  hasErrors(): boolean {
    return this.errors.length > 0;
  }

  getErrors(): Error[] {
    return [...this.errors];
  }

  throw(): never {
    if (this.errors.length === 1) {
      throw this.errors[0];
    }
    throw new AggregateError(
      this.errors,
      `${this.errors.length} errors occurred`,
    );
  }
}

// Usage: Validate multiple fields
function validateUser(data: any): User {
  const errors = new ErrorCollector();

  if (!data.email) {
    errors.add(new ValidationError("Email is required"));
  } else if (!isValidEmail(data.email)) {
    errors.add(new ValidationError("Email is invalid"));
  }

  if (!data.name || data.name.length < 2) {
    errors.add(new ValidationError("Name must be at least 2 characters"));
  }

  if (!data.age || data.age < 18) {
    errors.add(new ValidationError("Age must be 18 or older"));
  }

  if (errors.hasErrors()) {
    errors.throw();
  }

  return data as User;
}

Pattern 3: Graceful Degradation

模式3:优雅降级

Provide fallback functionality when errors occur.
python
from typing import Optional, Callable, TypeVar

T = TypeVar('T')

def with_fallback(
    primary: Callable[[], T],
    fallback: Callable[[], T],
    log_error: bool = True
) -> T:
    """Try primary function, fall back to fallback on error."""
    try:
        return primary()
    except Exception as e:
        if log_error:
            logger.error(f"Primary function failed: {e}")
        return fallback()
在发生错误时提供备用功能。
python
from typing import Optional, Callable, TypeVar

T = TypeVar('T')

def with_fallback(
    primary: Callable[[], T],
    fallback: Callable[[], T],
    log_error: bool = True
) -> T:
    """Try primary function, fall back to fallback on error."""
    try:
        return primary()
    except Exception as e:
        if log_error:
            logger.error(f"Primary function failed: {e}")
        return fallback()

Usage

Usage

def get_user_profile(user_id: str) -> UserProfile: return with_fallback( primary=lambda: fetch_from_cache(user_id), fallback=lambda: fetch_from_database(user_id) )
def get_user_profile(user_id: str) -> UserProfile: return with_fallback( primary=lambda: fetch_from_cache(user_id), fallback=lambda: fetch_from_database(user_id) )

Multiple fallbacks

Multiple fallbacks

def get_exchange_rate(currency: str) -> float: return ( try_function(lambda: api_provider_1.get_rate(currency)) or try_function(lambda: api_provider_2.get_rate(currency)) or try_function(lambda: cache.get_rate(currency)) or DEFAULT_RATE )
def try_function(func: Callable[[], Optional[T]]) -> Optional[T]: try: return func() except Exception: return None
undefined
def get_exchange_rate(currency: str) -> float: return ( try_function(lambda: api_provider_1.get_rate(currency)) or try_function(lambda: api_provider_2.get_rate(currency)) or try_function(lambda: cache.get_rate(currency)) or DEFAULT_RATE )
def try_function(func: Callable[[], Optional[T]]) -> Optional[T]: try: return func() except Exception: return None
undefined

Best Practices

最佳实践

  1. Fail Fast: Validate input early, fail quickly
  2. Preserve Context: Include stack traces, metadata, timestamps
  3. Meaningful Messages: Explain what happened and how to fix it
  4. Log Appropriately: Error = log, expected failure = don't spam logs
  5. Handle at Right Level: Catch where you can meaningfully handle
  6. Clean Up Resources: Use try-finally, context managers, defer
  7. Don't Swallow Errors: Log or re-throw, don't silently ignore
  8. Type-Safe Errors: Use typed errors when possible
python
undefined
  1. 快速失败:尽早验证输入,快速失败
  2. 保留上下文:包含栈跟踪、元数据、时间戳
  3. 有意义的提示:说明发生了什么以及如何修复
  4. 合理日志:错误=记录日志,预期内的失败=不要刷屏日志
  5. 在正确层级处理:在能够有效处理的层级捕获错误
  6. 清理资源:使用try-finally、上下文管理器、defer
  7. 不要掩盖错误:记录或重新抛出,不要静默忽略
  8. 类型安全错误:尽可能使用类型化错误
python
undefined

Good error handling example

Good error handling example

def process_order(order_id: str) -> Order: """Process order with comprehensive error handling.""" try: # Validate input if not order_id: raise ValidationError("Order ID is required")
    # Fetch order
    order = db.get_order(order_id)
    if not order:
        raise NotFoundError("Order", order_id)

    # Process payment
    try:
        payment_result = payment_service.charge(order.total)
    except PaymentServiceError as e:
        # Log and wrap external service error
        logger.error(f"Payment failed for order {order_id}: {e}")
        raise ExternalServiceError(
            f"Payment processing failed",
            service="payment_service",
            details={"order_id": order_id, "amount": order.total}
        ) from e

    # Update order
    order.status = "completed"
    order.payment_id = payment_result.id
    db.save(order)

    return order

except ApplicationError:
    # Re-raise known application errors
    raise
except Exception as e:
    # Log unexpected errors
    logger.exception(f"Unexpected error processing order {order_id}")
    raise ApplicationError(
        "Order processing failed",
        code="INTERNAL_ERROR"
    ) from e
undefined
def process_order(order_id: str) -> Order: """Process order with comprehensive error handling.""" try: # Validate input if not order_id: raise ValidationError("Order ID is required")
    # Fetch order
    order = db.get_order(order_id)
    if not order:
        raise NotFoundError("Order", order_id)

    # Process payment
    try:
        payment_result = payment_service.charge(order.total)
    except PaymentServiceError as e:
        # Log and wrap external service error
        logger.error(f"Payment failed for order {order_id}: {e}")
        raise ExternalServiceError(
            f"Payment processing failed",
            service="payment_service",
            details={"order_id": order_id, "amount": order.total}
        ) from e

    # Update order
    order.status = "completed"
    order.payment_id = payment_result.id
    db.save(order)

    return order

except ApplicationError:
    # Re-raise known application errors
    raise
except Exception as e:
    # Log unexpected errors
    logger.exception(f"Unexpected error processing order {order_id}")
    raise ApplicationError(
        "Order processing failed",
        code="INTERNAL_ERROR"
    ) from e
undefined

Common Pitfalls

常见陷阱

  • Catching Too Broadly:
    except Exception
    hides bugs
  • Empty Catch Blocks: Silently swallowing errors
  • Logging and Re-throwing: Creates duplicate log entries
  • Not Cleaning Up: Forgetting to close files, connections
  • Poor Error Messages: "Error occurred" is not helpful
  • Returning Error Codes: Use exceptions or Result types
  • Ignoring Async Errors: Unhandled promise rejections
  • 捕获范围过广
    except Exception
    会隐藏漏洞
  • 空捕获块:静默掩盖错误
  • 记录并重新抛出:创建重复日志条目
  • 未清理资源:忘记关闭文件、连接
  • 糟糕的错误提示:"Error occurred"毫无帮助
  • 返回错误码:使用异常或Result类型替代
  • 忽略异步错误:未处理的Promise拒绝

Resources

资源

  • references/exception-hierarchy-design.md: Designing error class hierarchies
  • references/error-recovery-strategies.md: Recovery patterns for different scenarios
  • references/async-error-handling.md: Handling errors in concurrent code
  • assets/error-handling-checklist.md: Review checklist for error handling
  • assets/error-message-guide.md: Writing helpful error messages
  • scripts/error-analyzer.py: Analyze error patterns in logs
  • references/exception-hierarchy-design.md:设计错误类层级
  • references/error-recovery-strategies.md:不同场景的恢复模式
  • references/async-error-handling.md:处理并发代码中的错误
  • assets/error-handling-checklist.md:错误处理审查清单
  • assets/error-message-guide.md:编写有用的错误提示
  • scripts/error-analyzer.py:分析日志中的错误模式