2.3 תבניות התנהגות פתרון

תרגיל 1 - Observer - פתרון

from abc import ABC, abstractmethod
from dataclasses import dataclass


@dataclass
class AssignmentEvent:
    task_id: int
    assignee_id: int
    assigner_id: int
    task_title: str


class AssignmentListener(ABC):
    @abstractmethod
    def on_assignment(self, event: AssignmentEvent):
        pass


class AssignmentEventBus:
    def __init__(self):
        self._listeners: list[AssignmentListener] = []

    def subscribe(self, listener: AssignmentListener):
        self._listeners.append(listener)

    def publish(self, event: AssignmentEvent):
        for listener in self._listeners:
            listener.on_assignment(event)


class EmailAssignmentListener(AssignmentListener):
    def on_assignment(self, event: AssignmentEvent):
        print(
            f"[EMAIL] למשתמש {event.assignee_id}: "
            f"הוקצתה לך משימה '{event.task_title}' על ידי {event.assigner_id}"
        )


class InAppNotificationListener(AssignmentListener):
    def on_assignment(self, event: AssignmentEvent):
        print(
            f"[IN-APP] התראה חדשה למשתמש {event.assignee_id}: "
            f"משימה '{event.task_title}' הוקצתה לך"
        )


class TaskService:
    def __init__(self, event_bus: AssignmentEventBus):
        self._event_bus = event_bus
        self._tasks = {
            1: {"id": 1, "title": "תקן באג", "assignee": None},
            2: {"id": 2, "title": "כתוב בדיקות", "assignee": None}
        }

    def assign_task(self, task_id: int, assignee_id: int, assigner_id: int):
        task = self._tasks.get(task_id)
        if not task:
            raise ValueError(f"משימה {task_id} לא נמצאה")
        task["assignee"] = assignee_id
        self._event_bus.publish(AssignmentEvent(
            task_id=task_id,
            assignee_id=assignee_id,
            assigner_id=assigner_id,
            task_title=task["title"]
        ))


bus = AssignmentEventBus()
bus.subscribe(EmailAssignmentListener())
bus.subscribe(InAppNotificationListener())

service = TaskService(bus)
service.assign_task(1, assignee_id=5, assigner_id=1)

---

תרגיל 2 - Strategy - פתרון

from abc import ABC, abstractmethod
from datetime import date


class TaskFilterStrategy(ABC):
    @abstractmethod
    def filter(self, tasks: list[dict]) -> list[dict]:
        pass


class FilterByStatus(TaskFilterStrategy):
    def __init__(self, status: str):
        self.status = status

    def filter(self, tasks: list[dict]) -> list[dict]:
        return [t for t in tasks if t.get("status") == self.status]


class FilterByOwner(TaskFilterStrategy):
    def __init__(self, owner_id: int):
        self.owner_id = owner_id

    def filter(self, tasks: list[dict]) -> list[dict]:
        return [t for t in tasks if t.get("owner_id") == self.owner_id]


class FilterOverdue(TaskFilterStrategy):
    def filter(self, tasks: list[dict]) -> list[dict]:
        today = date.today().isoformat()
        return [
            t for t in tasks
            if t.get("due_date") and t["due_date"] < today and t.get("status") != "done"
        ]


class CompositeFilter(TaskFilterStrategy):
    def __init__(self, *filters: TaskFilterStrategy):
        self._filters = list(filters)

    def filter(self, tasks: list[dict]) -> list[dict]:
        result = tasks
        for f in self._filters:
            result = f.filter(result)
        return result


# שימוש
tasks = [
    {"id": 1, "title": "תקן באג", "status": "in_progress", "owner_id": 1, "due_date": "2026-01-01"},
    {"id": 2, "title": "כתוב בדיקות", "status": "todo", "owner_id": 2, "due_date": "2026-12-31"},
    {"id": 3, "title": "review קוד", "status": "todo", "owner_id": 1, "due_date": "2026-01-15"},
]

# מסנן משימות פתוחות של user 1 שעברו תאריך יעד
f = CompositeFilter(FilterByOwner(1), FilterOverdue())
print(f.filter(tasks))

---

תרגיל 3 - Command - פתרון

class DeleteTaskCommand(Command):
    def __init__(self, db: TaskDB, task_id: int):
        self.db = db
        self.task_id = task_id
        self._deleted_task: dict | None = None

    def execute(self) -> None:
        self._deleted_task = self.db.delete(self.task_id)
        if not self._deleted_task:
            raise ValueError(f"משימה {self.task_id} לא נמצאה")

    def undo(self) -> None:
        if self._deleted_task:
            self.db._tasks[self._deleted_task["id"]] = self._deleted_task

    def description(self) -> str:
        title = self._deleted_task["title"] if self._deleted_task else f"id={self.task_id}"
        return f"מחיקת משימה: {title}"


# תרחיש
db = TaskDB()
history = CommandHistory()

# יצירת שתי משימות
history.execute(CreateTaskCommand(db, "משימה ראשונה", owner_id=1))
history.execute(CreateTaskCommand(db, "משימה שנייה", owner_id=1))

print("לפני מחיקה:", db._tasks)

# מחיקת המשימה הראשונה
history.execute(DeleteTaskCommand(db, 1))
print("אחרי מחיקה:", db._tasks)

# undo למחיקה
history.undo()
print("אחרי undo:", db._tasks)

# ודא שהמשימה חזרה
assert 1 in db._tasks, "המשימה הראשונה צריכה לחזור"
print("המשימה שוחזרה בהצלחה!")