Master cross-language error-handling patterns—including exceptions, Result types, error propagation, and graceful degradation—to build resilient applications. Use when implementing error handling, designing APIs, or improving application reliability.
English | 日本語
堅牢なエラーハンドリング戦略で、障害を適切に処理し、優れたデバッグ体験を提供する回復力のあるアプリケーションを構築します。
例外 vs Result型：
それぞれの使用タイミング：
回復可能なエラー：
回復不可能なエラー：
カスタム例外階層：
class ApplicationError(Exception):
"""すべてのアプリケーションエラーの基底例外。"""
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):
"""検証が失敗した時に発生。"""
pass
class NotFoundError(ApplicationError):
"""リソースが見つからない時に発生。"""
pass
class ExternalServiceError(ApplicationError):
"""外部サービスが失敗した時に発生。"""
def __init__(self, message: str, service: str, **kwargs):
super().__init__(message, **kwargs)
self.service = service
# 使用例
def get_user(user_id: str) -> User:
user = db.query(User).filter_by(id=user_id).first()
if not user:
raise NotFoundError(
f\"ユーザーが見つかりません\",
code=\"USER_NOT_FOUND\",
details={\"user_id\": user_id}
)
return user
クリーンアップ用のコンテキストマネージャー：
from contextlib import contextmanager
@contextmanager
def database_transaction(session):
"""トランザクションがコミットまたはロールバックされることを保証。"""
try:
yield session
session.commit()
except Exception as e:
session.rollback()
raise
finally:
session.close()
# 使用例
with database_transaction(db.session) as session:
user = User(name=\"Alice\")
session.add(user)
# 自動コミットまたはロールバック
指数バックオフ付きリトライ：
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,)
):
"""指数バックオフ付きリトライデコレーター。"""
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
# 使用例
@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()
カスタムエラークラス：
// カスタムエラークラス
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',
404,
{ resource, id }
);
}
}
// 使用例
function getUser(id: string): User {
const user = users.find(u => u.id === id);
if (!user) {
throw new NotFoundError('User', id);
}
return user;
}
Result型パターン：
// 明示的なエラーハンドリング用のResult型
type Result<T, E = Error> =
| { ok: true; value: T }
| { ok: false; error: E };
// ヘルパー関数
function Ok<T>(value: T): Result<T, never> {
return { ok: true, value };
}
function Err<E>(error: E): Result<never, E> {
return { ok: false, error };
}
// 使用例
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);
}
}
// Resultの消費
const result = parseJSON<User>(userJson);
if (result.ok) {
console.log(result.value.name);
} else {
console.error('パース失敗:', result.error.message);
}
// Resultのチェイン
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/await
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 []; // 見つからない場合は空配列を返す
}
if (error instanceof NetworkError) {
// リトライロジック
return retryFetchOrders(userId);
}
// 予期しないエラーを再スロー
throw error;
}
}
// Promiseエラーハンドリング
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('フェッチ失敗:', error);
throw error;
});
}
ResultとOption型：
use std::fs::File;
use std::io::{self, Read};
// 失敗する可能性のある操作用のResult型
fn read_file(path: &str) -> Result<String, io::Error> {
let mut file = File::open(path)?; // ?演算子でエラーを伝播
let mut contents = String::new();
file.read_to_string(&mut contents)?;
Ok(contents)
}
// カスタムエラー型
#[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)
}
}
// カスタムエラー型の使用
fn read_number_from_file(path: &str) -> Result<i32, AppError> {
let contents = read_file(path)?; // io::Errorを自動変換
let number = contents.trim().parse()
.map_err(AppError::Parse)?; // ParseIntErrorを明示的に変換
Ok(number)
}
// null許容値用のOption
fn find_user(id: &str) -> Option<User> {
users.iter().find(|u| u.id == id).cloned()
}
// Optionと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)
}
明示的なエラー戻り値：
// 基本的なエラーハンドリング
func getUser(id string) (*User, error) {
user, err := db.QueryUser(id)
if err != nil {
return nil, fmt.Errorf(\"ユーザークエリ失敗: %w\", err)
}
if user == nil {
return nil, errors.New(\"ユーザーが見つかりません\")
}
return user, nil
}
// カスタムエラー型
type ValidationError struct {
Field string
Message string
}
func (e *ValidationError) Error() string {
return fmt.Sprintf(\"%sの検証失敗: %s\", e.Field, e.Message)
}
// 比較用のセンチネルエラー
var (
ErrNotFound = errors.New(\"見つかりません\")
ErrUnauthorized = errors.New(\"権限がありません\")
ErrInvalidInput = errors.New(\"無効な入力\")
)
// エラーチェック
user, err := getUser(\"123\")
if err != nil {
if errors.Is(err, ErrNotFound) {
// 見つからない場合の処理
} else {
// その他のエラー処理
}
}
// エラーのラップとアンラップ
func processUser(id string) error {
user, err := getUser(id)
if err != nil {
return fmt.Errorf(\"ユーザー処理失敗: %w\", err)
}
// ユーザーを処理
return nil
}
// エラーのアンラップ
err := processUser(\"123\")
if err != nil {
var valErr *ValidationError
if errors.As(err, &valErr) {
fmt.Printf(\"検証エラー: %s\\n\", valErr.Field)
}
}
分散システムでのカスケード障害を防止。
from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, TypeVar
T = TypeVar('T')
class CircuitState(Enum):
CLOSED = \"closed\" # 正常動作
OPEN = \"open\" # 失敗中、リクエストを拒否
HALF_OPEN = \"half_open\" # 回復したかテスト中
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(\"サーキットブレーカーが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
# 使用例
circuit_breaker = CircuitBreaker()
def fetch_data():
return circuit_breaker.call(lambda: external_api.get_data())
最初のエラーで失敗するのではなく、複数のエラーを収集。
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}個のエラーが発生しました`
);
}
}
// 使用例：複数フィールドの検証
function validateUser(data: any): User {
const errors = new ErrorCollector();
if (!data.email) {
errors.add(new ValidationError('メールアドレスは必須です'));
} else if (!isValidEmail(data.email)) {
errors.add(new ValidationError('メールアドレスが無効です'));
}
if (!data.name || data.name.length < 2) {
errors.add(new ValidationError('名前は2文字以上である必要があります'));
}
if (!data.age || data.age < 18) {
errors.add(new ValidationError('年齢は18歳以上である必要があります'));
}
if (errors.hasErrors()) {
errors.throw();
}
return data as User;
}
エラー発生時にフォールバック機能を提供。
from typing import Optional, Callable, TypeVar
T = TypeVar('T')
def with_fallback(
primary: Callable[[], T],
fallback: Callable[[], T],
log_error: bool = True
) -> T:
"""プライマリ関数を試行し、エラー時はフォールバックに切り替え。"""
try:
return primary()
except Exception as e:
if log_error:
logger.error(f\"プライマリ関数が失敗: {e}\")
return fallback()
# 使用例
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_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
# 良いエラーハンドリングの例
def process_order(order_id: str) -> Order:
"""包括的なエラーハンドリング付きで注文を処理。"""
try:
# 入力を検証
if not order_id:
raise ValidationError(\"注文IDが必要です\")
# 注文を取得
order = db.get_order(order_id)
if not order:
raise NotFoundError(\"Order\", order_id)
# 決済を処理
try:
payment_result = payment_service.charge(order.total)
except PaymentServiceError as e:
# 外部サービスエラーをログに記録してラップ
logger.error(f\"注文{order_id}の決済失敗: {e}\")
raise ExternalServiceError(
f\"決済処理に失敗しました\",
service=\"payment_service\",
details={\"order_id\": order_id, \"amount\": order.total}
) from e
# 注文を更新
order.status = \"completed\"
order.payment_id = payment_result.id
db.save(order)
return order
except ApplicationError:
# 既知のアプリケーションエラーを再スロー
raise
except Exception as e:
# 予期しないエラーをログ
logger.exception(f\"注文{order_id}の処理で予期しないエラー\")
raise ApplicationError(
\"注文処理に失敗しました\",
code=\"INTERNAL_ERROR\"
) from e
except Exceptionはバグを隠すnpx skills add amurata/error-handling-patterns下载完整 Skill 目录，包含 SKILL.md 及所有相关文件
Search for places (restaurants, cafes, etc.) via Google Places API proxy on localhost.
Interact with GitHub using the `gh` CLI. Use `gh issue`, `gh pr`, `gh run`, and `gh api` for issues, PRs, CI runs, and advanced queries.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Start voice calls via the OpenClaw voice-call plugin.
Notion API for creating and managing pages, databases, and blocks.
Gemini CLI for one-shot Q&A, summaries, and generation.
Category:developer