2.2 - תבניות מבנה - פתרון

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

class FileStorageAdapter(AttachmentStore):
    FOLDER = "attachments"

    def __init__(self, storage: FileStorage):
        self._storage = storage

    def save(self, content: bytes, name: str) -> str:
        # Adapts save -> upload_file
        url = self._storage.upload_file(content, name, self.FOLDER)
        return url  # the URL serves as the attachment_id

    def delete(self, attachment_id: str) -> None:
        # Adapts delete -> remove_file
        success = self._storage.remove_file(attachment_id)
        if not success:
            raise RuntimeError(f"Failed to delete: {attachment_id}")

    def get(self, attachment_id: str) -> bytes:
        # Adapts get -> retrieve_file
        return self._storage.retrieve_file(attachment_id)


# Usage
storage = FileStorage()
adapter = FileStorageAdapter(storage)

attachment_id = adapter.save(b"file content", "report.pdf")
content = adapter.get(attachment_id)
adapter.delete(attachment_id)

---

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

import time


class TimedTaskService:
    def __init__(self, wrapped: TaskService):
        self._wrapped = wrapped

    def get_task(self, task_id: int) -> dict:
        start = time.perf_counter()
        result = self._wrapped.get_task(task_id)
        elapsed = time.perf_counter() - start
        print(f"[TIMER] get_task took {elapsed:.3f}s")
        return result

    def update_task(self, task_id: int, title: str) -> dict:
        start = time.perf_counter()
        result = self._wrapped.update_task(task_id, title)
        elapsed = time.perf_counter() - start
        print(f"[TIMER] update_task took {elapsed:.3f}s")
        return result


class RetryTaskService:
    MAX_RETRIES = 3

    def __init__(self, wrapped: TaskService):
        self._wrapped = wrapped

    def get_task(self, task_id: int) -> dict:
        return self._wrapped.get_task(task_id)

    def update_task(self, task_id: int, title: str) -> dict:
        last_error = None
        for attempt in range(1, self.MAX_RETRIES + 1):
            try:
                return self._wrapped.update_task(task_id, title)
            except Exception as e:
                last_error = e
                print(f"[RETRY] attempt {attempt} failed: {e}")
        raise last_error


# Chaining decorators
base = TaskService()
with_retry = RetryTaskService(base)
with_timing_and_retry = TimedTaskService(with_retry)

result = with_timing_and_retry.update_task(1, "New title")

---

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

class SimpleTaskRepo:
    def __init__(self):
        self._tasks = []
        self._next_id = 1

    def create(self, title: str, description: str, owner_id: int) -> dict:
        task = {"id": self._next_id, "title": title, "description": description,
                "owner_id": owner_id, "status": "open", "project_id": None}
        self._tasks.append(task)
        self._next_id += 1
        return task

    def find_open_by_project(self, project_id: int) -> list[dict]:
        return [t for t in self._tasks if t["project_id"] == project_id and t["status"] == "open"]

    def mark_done(self, task_id: int):
        for task in self._tasks:
            if task["id"] == task_id:
                task["status"] = "done"


class SimpleProjectRepo:
    def __init__(self):
        self._projects = {}

    def archive(self, project_id: int):
        self._projects[project_id] = {"id": project_id, "archived": True}
        print(f"Project {project_id} archived")


class SimpleEmailService:
    def send(self, to: str, subject: str, body: str):
        print(f"Email to {to} | subject: {subject} | body: {body}")


class TaskFlowFacade:
    def __init__(self):
        self._tasks = SimpleTaskRepo()
        self._projects = SimpleProjectRepo()
        self._email = SimpleEmailService()

    def create_task_with_notification(
        self,
        title: str,
        description: str,
        owner_id: int,
        owner_email: str
    ) -> dict:
        task = self._tasks.create(title, description, owner_id)
        self._email.send(
            to=owner_email,
            subject=f"New task: {title}",
            body=f"A new task was created for you: {title}\n{description}"
        )
        return task

    def close_project(self, project_id: int, owner_id: int):
        open_tasks = self._tasks.find_open_by_project(project_id)
        for task in open_tasks:
            self._tasks.mark_done(task["id"])
        self._projects.archive(project_id)
        print(f"Project {project_id} closed. {len(open_tasks)} tasks marked as done.")


# Usage
facade = TaskFlowFacade()
task = facade.create_task_with_notification("Fix bug", "Bug in login", 1, "dev@example.com")
facade.close_project(project_id=1, owner_id=1)