class Retry: """失敗時にリトライするデコレータ""" def __init__(self, max_retries: int = 3): self.max_retries = max_retries def __call__(self, func): def wrapper(*args, **kwargs): last_exception = None for attempt in range(self.max_retries): try: return func(*args, **kwargs) except Exception as e: last_exception = e print(f"Attempt {attempt + 1} failed: {e}") raise last_exception return wrapper@Retry(max_retries=3)def unstable_api_call(): import random if random.random() < 0.7: raise ConnectionError("API connection failed") return "Success!"# 実行例# result = unstable_api_call()
実践的な使用パターン
動的な関数呼び出し
def safe_call(func, *args, **kwargs): """安全に関数を呼び出す""" if not callable(func): raise TypeError(f"{type(func).__name__} is not callable") return func(*args, **kwargs)# 使用例def add(a, b): return a + bresult = safe_call(add, 1, 2)print(result) # 3# エラーケースtry: safe_call("not a function", 1, 2)except TypeError as e: print(e) # str is not callable
プラグインシステム
class PluginManager: """プラグインを管理・実行するマネージャー""" def __init__(self): self.plugins = {} def register(self, name: str, plugin): """プラグインを登録(呼び出し可能なオブジェクトのみ)""" if not callable(plugin): raise TypeError(f"Plugin must be callable, got {type(plugin).__name__}") self.plugins[name] = plugin def execute(self, name: str, *args, **kwargs): """プラグインを実行""" if name not in self.plugins: raise KeyError(f"Plugin '{name}' not found") return self.plugins[name](*args, **kwargs) def list_plugins(self) -> list: """登録済みプラグインの一覧""" return list(self.plugins.keys())# プラグインの定義def uppercase_plugin(text: str) -> str: return text.upper()class ReversePlugin: def __call__(self, text: str) -> str: return text[::-1]# 使用例manager = PluginManager()manager.register("uppercase", uppercase_plugin)manager.register("reverse", ReversePlugin())print(manager.execute("uppercase", "hello")) # HELLOprint(manager.execute("reverse", "hello")) # ollehprint(manager.list_plugins()) # ['uppercase', 'reverse']
イベントハンドラシステム
from typing import Callable, Dict, List, Anyclass EventEmitter: """イベント駆動のハンドラ管理""" def __init__(self): self._handlers: Dict[str, List[Callable]] = {} def on(self, event: str, handler: Callable): """イベントハンドラを登録""" if not callable(handler): raise TypeError("Handler must be callable") if event not in self._handlers: self._handlers[event] = [] self._handlers[event].append(handler) def emit(self, event: str, *args, **kwargs) -> List[Any]: """イベントを発火""" results = [] if event in self._handlers: for handler in self._handlers[event]: results.append(handler(*args, **kwargs)) return results def off(self, event: str, handler: Callable = None): """ハンドラを解除""" if event in self._handlers: if handler is None: del self._handlers[event] else: self._handlers[event] = [ h for h in self._handlers[event] if h != handler ]# 使用例emitter = EventEmitter()def on_user_login(user_id: int): print(f"User {user_id} logged in")def send_welcome_email(user_id: int): print(f"Sending welcome email to user {user_id}")emitter.on("login", on_user_login)emitter.on("login", send_welcome_email)emitter.emit("login", 123)# 出力:# User 123 logged in# Sending welcome email to user 123