error-handling-patterns
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseError 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 = serviceUsage
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 decoratorwith 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 decoratorUsage
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()
undefinedTypeScript/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.OPENUsage
Usage
circuit_breaker = CircuitBreaker()
def fetch_data():
return circuit_breaker.call(lambda: external_api.get_data())
undefinedcircuit_breaker = CircuitBreaker()
def fetch_data():
return circuit_breaker.call(lambda: external_api.get_data())
undefinedPattern 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
undefineddef 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
undefinedBest Practices
最佳实践
- Fail Fast: Validate input early, fail quickly
- Preserve Context: Include stack traces, metadata, timestamps
- Meaningful Messages: Explain what happened and how to fix it
- Log Appropriately: Error = log, expected failure = don't spam logs
- Handle at Right Level: Catch where you can meaningfully handle
- Clean Up Resources: Use try-finally, context managers, defer
- Don't Swallow Errors: Log or re-throw, don't silently ignore
- Type-Safe Errors: Use typed errors when possible
python
undefined- 快速失败:尽早验证输入,快速失败
- 保留上下文:包含栈跟踪、元数据、时间戳
- 有意义的提示:说明发生了什么以及如何修复
- 合理日志:错误=记录日志,预期内的失败=不要刷屏日志
- 在正确层级处理:在能够有效处理的层级捕获错误
- 清理资源:使用try-finally、上下文管理器、defer
- 不要掩盖错误:记录或重新抛出,不要静默忽略
- 类型安全错误:尽可能使用类型化错误
python
undefinedGood 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 eundefineddef 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 eundefinedCommon Pitfalls
常见陷阱
- Catching Too Broadly: hides bugs
except Exception - 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:分析日志中的错误模式