timmy / Python Mixin 設計模式

import datetime import json import traceback import threading class LoggingMixin: """ 提供日誌功能的 Mixin 類別，可選擇寫入控制台或檔案 """ def __init__(self, log_to_file=False, log_file="app.log", *args, **kwargs): self.log_to_file = log_to_file self.log_file = log_file super().__init__(*args, **kwargs) def log(self, message): """ 記錄日誌 """ timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_message = f"[{timestamp}] {self.__class__.__name__}: {message}" if self.log_to_file: with open(self.log_file, "a") as f: f.write(log_message + "\n") else: print(log_message) class TimestampMixin: """ 提供時間標記功能，支援時區與 ISO 格式 """ def __init__(self, timezone="UTC", *args, **kwargs): self.timezone = timezone self.created_at = datetime.datetime.now(datetime.timezone.utc).isoformat() self.updated_at = self.created_at super().__init__(*args, **kwargs) def update_timestamp(self): """ 更新 updated_at 時間 """ self.updated_at = datetime.datetime.now(datetime.timezone.utc).isoformat() self.log(f"更新時間標記: {self.updated_at}") class DatabaseMixin: """ 提供資料儲存，支援 JSON 持久化、本機快取 """ _database = {} def save(self): """ 儲存物件至模擬資料庫 """ DatabaseMixin._database[self.name] = json.dumps(self.__dict__, default=str) self.log(f"資料已儲存: {self.name}") @classmethod def get(cls, name): """ 從資料庫中獲取物件 """ data = cls._database.get(name, None) return json.loads(data) if data else None class PermissionMixin: """ 提供基於角色的權限管理 (RBAC) """ _roles_permissions = { "admin": ["create", "read", "update", "delete"], "editor": ["create", "read", "update"], "viewer": ["read"] } def __init__(self, role="viewer", *args, **kwargs): self.role = role self.permissions = self._roles_permissions.get(role, []) super().__init__(*args, **kwargs) def has_permission(self, perm): """ 檢查是否擁有指定權限 """ return perm in self.permissions def change_role(self, new_role): """ 更新角色權限 """ self.role = new_role self.permissions = self._roles_permissions.get(new_role, []) self.log(f"角色變更為 {new_role}，新權限: {self.permissions}") class EventMixin: """ 提供同步與非同步事件監聽 """ _events = {} @classmethod def on(cls, event_name, callback, async_mode=False): """ 訂閱事件，支援同步或非同步 """ if event_name not in cls._events: cls._events[event_name] = [] cls._events[event_name].append((callback, async_mode)) @classmethod def trigger(cls, event_name, *args, **kwargs): """ 觸發事件 """ if event_name in cls._events: for callback, async_mode in cls._events[event_name]: if async_mode: threading.Thread(target=callback, args=args, kwargs=kwargs).start() else: callback(*args, **kwargs) class ExceptionHandlingMixin: """ 提供異常處理與回復策略 """ def handle_exception(self, func, *args, **kwargs): """ 捕獲異常並記錄 """ try: return func(*args, **kwargs) except Exception as e: self.log(f"發生異常: {e}") traceback.print_exc() class AuditMixin: """ 追蹤所有變更，提供完整審計歷史 """ _audit_log = [] def audit(self, action, details): """ 記錄審計日誌 """ record = { "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), "user": self.name, "action": action, "details": details } self._audit_log.append(record) self.log(f"審計記錄: {record}") class User(LoggingMixin, TimestampMixin, DatabaseMixin, PermissionMixin, EventMixin, ExceptionHandlingMixin, AuditMixin): """ Grandmaster User 類別，擁有完整的日誌、權限、事件、異常處理與審計功能 """ def __init__(self, name, role="viewer", log_to_file=False): super().__init__(role=role, log_to_file=log_to_file) self.name = name self.log(f"User '{self.name}' 已建立，角色: {self.role}") def update_name(self, new_name): """ 更新使用者名稱，並觸發事件與審計記錄 """ self.audit("update_name", {"old_name": self.name, "new_name": new_name}) self.log(f"名稱從 '{self.name}' 變更為 '{new_name}'") old_name = self.name self.name = new_name self.update_timestamp() self.trigger("user_updated", old_name, new_name) # 測試範例 if __name__ == "__main__": # 訂閱事件（同步與非同步） def on_user_updated(old_name, new_name): print(f"🔔 使用者名稱變更通知: {old_name} → {new_name}") User.on("user_updated", on_user_updated) User.on("user_updated", lambda o, n: print(f"📡 非同步通知: {o} → {n}"), async_mode=True) # 建立使用者 user1 = User("Alice", role="editor", log_to_file=True) user1.save() # 儲存到模擬資料庫 # 更新名稱 user1.update_name("Alice_Wonderland") # 檢查權限 print(f"Alice 是否有 'read' 權限？{user1.has_permission('read')}") print(f"Alice 是否有 'delete' 權限？{user1.has_permission('delete')}") # 變更角色 user1.change_role("admin") # 從資料庫取得使用者 retrieved_user = User.get("Alice_Wonderland") if retrieved_user: print(f"從資料庫取得使用者: {retrieved_user}") # 測試異常處理 user1.handle_exception(lambda: 1 / 0) # 觸發除零錯誤