2.3 - תבניות התנהגות - הרצאה
תבניות התנהגות - Behavioral Patterns¶
תבניות התנהגות עוסקות בתקשורת ואחריות בין אובייקטים - מי מחליט מה, ואיך אובייקטים מגיבים לevents.
נכיר שלוש תבניות: Observer, Strategy ו-Command.
Observer - משקיף¶
הבעיה: כשמשהו קורה באובייקט אחד, אובייקטים אחרים צריכים לדעת ולהגיב - אבל אנחנו לא רוצים שהאובייקט הראשון יידע מי מקשיב לו.
הפתרון: האובייקט "מפרסם" events, והמעוניינים "נרשמים" לevents שמעניינים אותם.
בלי Observer - בעיתי¶
class TaskService:
def complete_task(self, task_id: int):
task = self._repo.find_by_id(task_id)
task["status"] = "done"
self._repo.save(task)
# TaskService knows everyone who needs to be notified - a problem!
EmailService().send_completion_email(task)
SlackService().notify_channel(task)
AuditLogger().log_completion(task)
StatisticsService().increment_completed(task)
אם רוצים להוסיף שירות חדש, צריך לשנות את TaskService. הוא יודע יותר מדי.
עם Observer¶
from abc import ABC, abstractmethod
class TaskEvent:
def __init__(self, event_type: str, task: dict):
self.event_type = event_type
self.task = task
class TaskEventListener(ABC):
@abstractmethod
def on_task_event(self, event: TaskEvent):
pass
class EventBus:
def __init__(self):
self._listeners: dict[str, list[TaskEventListener]] = {}
def subscribe(self, event_type: str, listener: TaskEventListener):
if event_type not in self._listeners:
self._listeners[event_type] = []
self._listeners[event_type].append(listener)
def publish(self, event: TaskEvent):
for listener in self._listeners.get(event.event_type, []):
listener.on_task_event(event)
class EmailNotificationListener(TaskEventListener):
def on_task_event(self, event: TaskEvent):
if event.event_type == "task_completed":
print(f"Sending email: task '{event.task['title']}' completed")
class SlackNotificationListener(TaskEventListener):
def on_task_event(self, event: TaskEvent):
print(f"Slack: task '{event.task['title']}' changed - {event.event_type}")
class AuditLogListener(TaskEventListener):
def on_task_event(self, event: TaskEvent):
print(f"[AUDIT] {event.event_type}: {event.task}")
class TaskService:
def __init__(self, event_bus: EventBus):
self._event_bus = event_bus
self._tasks = {}
def complete_task(self, task_id: int):
task = self._tasks.get(task_id, {"id": task_id, "title": f"Task {task_id}"})
task["status"] = "done"
# TaskService doesn't know who's listening - it just publishes
self._event_bus.publish(TaskEvent("task_completed", task))
# Wiring everything together
bus = EventBus()
bus.subscribe("task_completed", EmailNotificationListener())
bus.subscribe("task_completed", SlackNotificationListener())
bus.subscribe("task_completed", AuditLogListener())
service = TaskService(bus)
service.complete_task(42)
עכשיו אפשר להוסיף StatisticsListener בלי לשנות שורה אחת ב-TaskService.
Strategy - אסטרטגיה¶
הבעיה: יש לנו אלגוריתם שמשתנה בהתאם להקשר. אנחנו לא רוצים ערמות של if/elif.
הפתרון: הגדרת "משפחה" של אלגוריתמים שניתן להחלפה בזמן ריצה.
דוגמה: מיון משימות¶
from abc import ABC, abstractmethod
class TaskSortStrategy(ABC):
@abstractmethod
def sort(self, tasks: list[dict]) -> list[dict]:
pass
class SortByDueDate(TaskSortStrategy):
def sort(self, tasks: list[dict]) -> list[dict]:
return sorted(tasks, key=lambda t: t.get("due_date") or "9999-12-31")
class SortByPriority(TaskSortStrategy):
PRIORITY_ORDER = {"critical": 0, "high": 1, "normal": 2, "low": 3}
def sort(self, tasks: list[dict]) -> list[dict]:
return sorted(tasks, key=lambda t: self.PRIORITY_ORDER.get(t.get("priority", "normal"), 2))
class SortByTitle(TaskSortStrategy):
def sort(self, tasks: list[dict]) -> list[dict]:
return sorted(tasks, key=lambda t: t.get("title", ""))
class TaskList:
def __init__(self, tasks: list[dict]):
self._tasks = tasks
self._sort_strategy: TaskSortStrategy = SortByDueDate()
def set_sort_strategy(self, strategy: TaskSortStrategy):
self._sort_strategy = strategy
def get_sorted_tasks(self) -> list[dict]:
return self._sort_strategy.sort(self._tasks)
# usage
tasks = [
{"title": "Fix bug", "priority": "high", "due_date": "2026-04-10"},
{"title": "Write documentation", "priority": "low", "due_date": "2026-04-05"},
{"title": "review code", "priority": "normal", "due_date": "2026-04-08"},
]
task_list = TaskList(tasks)
task_list.set_sort_strategy(SortByDueDate())
print("By date:", [t["title"] for t in task_list.get_sorted_tasks()])
task_list.set_sort_strategy(SortByPriority())
print("By priority:", [t["title"] for t in task_list.get_sorted_tasks()])
task_list.set_sort_strategy(SortByTitle())
print("By title:", [t["title"] for t in task_list.get_sorted_tasks()])
Command - פקודה¶
הבעיה: רוצים לעטוף פעולות כאובייקטים - כדי לדחות ביצוע, לתור אותן, לבטל אותן (undo), או לתעד אותן.
הפתרון: כל פעולה הופכת לאובייקט עם execute() ו-undo().
דוגמה: undo/redo במערכת משימות¶
from abc import ABC, abstractmethod
class Command(ABC):
@abstractmethod
def execute(self) -> None:
pass
@abstractmethod
def undo(self) -> None:
pass
@abstractmethod
def description(self) -> str:
pass
class TaskDB:
"""Simulates a simple database"""
def __init__(self):
self._tasks: dict[int, dict] = {}
self._next_id = 1
def insert(self, task: dict) -> int:
task_id = self._next_id
self._tasks[task_id] = {**task, "id": task_id}
self._next_id += 1
return task_id
def delete(self, task_id: int) -> dict:
return self._tasks.pop(task_id, None)
def update(self, task_id: int, **fields):
if task_id in self._tasks:
self._tasks[task_id].update(fields)
def get(self, task_id: int) -> dict | None:
return self._tasks.get(task_id)
class CreateTaskCommand(Command):
def __init__(self, db: TaskDB, title: str, owner_id: int):
self.db = db
self.title = title
self.owner_id = owner_id
self._created_id: int | None = None
def execute(self) -> None:
self._created_id = self.db.insert({"title": self.title, "owner_id": self.owner_id})
def undo(self) -> None:
if self._created_id:
self.db.delete(self._created_id)
def description(self) -> str:
return f"Creating task: {self.title}"
class UpdateTaskStatusCommand(Command):
def __init__(self, db: TaskDB, task_id: int, new_status: str):
self.db = db
self.task_id = task_id
self.new_status = new_status
self._old_status: str | None = None
def execute(self) -> None:
task = self.db.get(self.task_id)
if task:
self._old_status = task.get("status")
self.db.update(self.task_id, status=self.new_status)
def undo(self) -> None:
if self._old_status is not None:
self.db.update(self.task_id, status=self._old_status)
def description(self) -> str:
return f"Updating status of task {self.task_id} to {self.new_status}"
class CommandHistory:
def __init__(self):
self._history: list[Command] = []
def execute(self, command: Command) -> None:
command.execute()
self._history.append(command)
print(f"Executed: {command.description()}")
def undo(self) -> None:
if not self._history:
print("No actions to undo")
return
command = self._history.pop()
command.undo()
print(f"Undone: {command.description()}")
def history(self) -> list[str]:
return [cmd.description() for cmd in self._history]
# usage
db = TaskDB()
history = CommandHistory()
history.execute(CreateTaskCommand(db, "Fix bug", owner_id=1))
history.execute(UpdateTaskStatusCommand(db, 1, "in_progress"))
history.execute(UpdateTaskStatusCommand(db, 1, "done"))
print("History:", history.history())
history.undo() # undoing "done"
history.undo() # undoing "in_progress"
print("History after undo:", history.history())
סיכום¶
| תבנית | מה היא פותרת |
|---|---|
| Observer | תקשורת בין אובייקטים בלי תלות ישירה |
| Strategy | החלפת אלגוריתמים בזמן ריצה |
| Command | עטיפת פעולות כאובייקטים לצורך undo/queue/log |