timmy / Python Mixin 設計模式

import datetime import json import threading import traceback class LoggingMixin: 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: def __init__(self, *args, **kwargs): self.created_at = datetime.datetime.now().isoformat() self.updated_at = self.created_at super().__init__(*args, **kwargs) def update_timestamp(self): self.updated_at = datetime.datetime.now().isoformat() self.log(f"更新時間標記: {self.updated_at}") class DatabaseMixin: _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: _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().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): 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)