2.3 תבניות התנהגות הרצאה

תבניות התנהגות - Behavioral Patterns

תבניות התנהגות עוסקות בתקשורת ואחריות בין אובייקטים - מי מחליט מה, ואיך אובייקטים מגיבים לאירועים.

נכיר שלוש תבניות: Observer, Strategy ו-Command.

---

Observer - משקיף

הבעיה: כשמשהו קורה באובייקט אחד, אובייקטים אחרים צריכים לדעת ולהגיב - אבל אנחנו לא רוצים שהאובייקט הראשון יידע מי מקשיב לו.

הפתרון: האובייקט "מפרסם" אירועים, והמעוניינים "נרשמים" לאירועים שמעניינים אותם.

בלי 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 מכיר את כל מי שצריך להודיע לו - בעיה!
        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"שולח מייל: משימה '{event.task['title']}' הושלמה")


class SlackNotificationListener(TaskEventListener):
    def on_task_event(self, event: TaskEvent):
        print(f"Slack: משימה '{event.task['title']}' שונתה - {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_id}"})
        task["status"] = "done"
        # TaskService לא יודע מי מקשיב - רק מפרסם
        self._event_bus.publish(TaskEvent("task_completed", task))


# חיבור הכל יחד
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)


# שימוש
tasks = [
    {"title": "תקן באג", "priority": "high", "due_date": "2026-04-10"},
    {"title": "כתוב תיעוד", "priority": "low", "due_date": "2026-04-05"},
    {"title": "review קוד", "priority": "normal", "due_date": "2026-04-08"},
]

task_list = TaskList(tasks)

task_list.set_sort_strategy(SortByDueDate())
print("לפי תאריך:", [t["title"] for t in task_list.get_sorted_tasks()])

task_list.set_sort_strategy(SortByPriority())
print("לפי עדיפות:", [t["title"] for t in task_list.get_sorted_tasks()])

task_list.set_sort_strategy(SortByTitle())
print("לפי כותרת:", [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:
    """מדמה מסד נתונים פשוט"""
    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"יצירת משימה: {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"עדכון סטטוס משימה {self.task_id} ל-{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"בוצע: {command.description()}")

    def undo(self) -> None:
        if not self._history:
            print("אין פעולות לביטול")
            return
        command = self._history.pop()
        command.undo()
        print(f"בוטל: {command.description()}")

    def history(self) -> list[str]:
        return [cmd.description() for cmd in self._history]


# שימוש
db = TaskDB()
history = CommandHistory()

history.execute(CreateTaskCommand(db, "תקן באג", owner_id=1))
history.execute(UpdateTaskStatusCommand(db, 1, "in_progress"))
history.execute(UpdateTaskStatusCommand(db, 1, "done"))

print("היסטוריה:", history.history())

history.undo()  # מבטל "done"
history.undo()  # מבטל "in_progress"

print("היסטוריה אחרי undo:", history.history())

סיכום

תבנית	מה היא פותרת
Observer	תקשורת בין אובייקטים בלי תלות ישירה
Strategy	החלפת אלגוריתמים בזמן ריצה
Command	עטיפת פעולות כאובייקטים לצורך undo/queue/log