1.3 ארכיטקטורה נקייה הרצאה

ארכיטקטורה נקייה - Clean Architecture

ארכיטקטורה נקייה היא גישה שפותחה על ידי Robert C. Martin, ומטרתה ליצור מערכות שבהן:
- הלוגיקה העסקית נפרדת לחלוטין מ-frameworks, מסדי נתונים וממשק המשתמש
- ניתן להחליף את ה-framework, מסד הנתונים או ה-UI בלי לשנות את הלוגיקה
- ניתן לבדוק את הלוגיקה העסקית בלי להריץ שרת, מסד נתונים או HTTP

ארבע השכבות

+------------------------------------------+
|        Frameworks & Drivers              |  (FastAPI, SQLite, Redis)
|  +------------------------------------+  |
|  |      Interface Adapters            |  |  (Routes, Repositories)
|  |  +--------------------------+  |  |
|  |  |      Use Cases           |  |  |  (Application Logic)
|  |  |  +------------------+  |  |  |
|  |  |  |    Entities      |  |  |  |  (Core Business Objects)
|  |  |  +------------------+  |  |  |
|  |  +--------------------------+  |  |
|  +------------------------------------+  |
+------------------------------------------+

כלל הבסיס: התלויות תמיד מצביעות פנימה. שכבה חיצונית יכולה לתלות בשכבה פנימית, אבל לא להפך.

---

שכבה 1: Entities - ישויות

הישויות הן אובייקטי הליבה של הביזנס - הם לא יודעים כלום על FastAPI, SQLite, HTTP או כל דבר טכנולוגי.

# entities/task.py
from dataclasses import dataclass
from datetime import datetime
from enum import Enum


class TaskStatus(Enum):
    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"


@dataclass
class Task:
    id: int
    title: str
    description: str
    status: TaskStatus
    owner_id: int
    created_at: datetime
    due_date: datetime | None = None

    def is_overdue(self) -> bool:
        if self.due_date is None or self.status == TaskStatus.DONE:
            return False
        return datetime.now() > self.due_date

    def complete(self):
        self.status = TaskStatus.DONE

שימו לב: הקלאס Task הוא פייתון טהור. אין imports של FastAPI, SQLAlchemy, או כל ספרייה חיצונית.

---

שכבה 2: Use Cases - מקרי שימוש

מקרי שימוש הם הלוגיקה הספציפית לאפליקציה. הם מתארים מה האפליקציה עושה.

# use_cases/task_use_cases.py
from entities.task import Task, TaskStatus
from datetime import datetime


class TaskRepository:
    """ממשק מופשט - לא יודע איזה DB"""
    def save(self, task: Task) -> Task: ...
    def find_by_id(self, task_id: int) -> Task | None: ...
    def find_by_owner(self, owner_id: int) -> list[Task]: ...


class NotificationPort:
    """ממשק מופשט - לא יודע איך שולחים"""
    def notify_task_created(self, task: Task, owner_email: str): ...


class CreateTaskUseCase:
    def __init__(self, repository: TaskRepository, notifier: NotificationPort):
        self.repository = repository
        self.notifier = notifier

    def execute(self, title: str, description: str, owner_id: int, owner_email: str) -> Task:
        if len(title) < 3:
            raise ValueError("כותרת קצרה מדי")
        task = Task(
            id=0,
            title=title,
            description=description,
            status=TaskStatus.TODO,
            owner_id=owner_id,
            created_at=datetime.now()
        )
        saved_task = self.repository.save(task)
        self.notifier.notify_task_created(saved_task, owner_email)
        return saved_task


class CompleteTaskUseCase:
    def __init__(self, repository: TaskRepository):
        self.repository = repository

    def execute(self, task_id: int, requesting_user_id: int) -> Task:
        task = self.repository.find_by_id(task_id)
        if task is None:
            raise ValueError(f"משימה {task_id} לא נמצאה")
        if task.owner_id != requesting_user_id:
            raise PermissionError("רק הבעלים יכול להשלים את המשימה")
        task.complete()
        return self.repository.save(task)

---

שכבה 3: Interface Adapters - מתאמי ממשק

שכבה זו ממירה נתונים בין הפורמט שמתאים ל-use cases לפורמט שמתאים לכלים החיצוניים.

# adapters/sqlite_task_repository.py
import sqlite3
from entities.task import Task, TaskStatus
from use_cases.task_use_cases import TaskRepository
from datetime import datetime


class SQLiteTaskRepository(TaskRepository):
    def __init__(self, db_path: str):
        self.db_path = db_path

    def save(self, task: Task) -> Task:
        db = sqlite3.connect(self.db_path)
        if task.id == 0:
            cursor = db.execute(
                "INSERT INTO tasks (title, description, status, owner_id, created_at) VALUES (?, ?, ?, ?, ?)",
                (task.title, task.description, task.status.value, task.owner_id, task.created_at.isoformat())
            )
            db.commit()
            task.id = cursor.lastrowid
        else:
            db.execute(
                "UPDATE tasks SET status=? WHERE id=?",
                (task.status.value, task.id)
            )
            db.commit()
        return task

    def find_by_id(self, task_id: int) -> Task | None:
        db = sqlite3.connect(self.db_path)
        row = db.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
        if row is None:
            return None
        return Task(
            id=row[0], title=row[1], description=row[2],
            status=TaskStatus(row[3]), owner_id=row[4],
            created_at=datetime.fromisoformat(row[5])
        )

    def find_by_owner(self, owner_id: int) -> list[Task]:
        db = sqlite3.connect(self.db_path)
        rows = db.execute("SELECT * FROM tasks WHERE owner_id=?", (owner_id,)).fetchall()
        return [self.find_by_id(row[0]) for row in rows]

---

שכבה 4: Frameworks & Drivers - FastAPI

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from use_cases.task_use_cases import CreateTaskUseCase, CompleteTaskUseCase
from adapters.sqlite_task_repository import SQLiteTaskRepository
from adapters.email_notifier import EmailNotifier

app = FastAPI()

# הזרקת התלויות
repository = SQLiteTaskRepository("taskflow.db")
notifier = EmailNotifier()
create_task_use_case = CreateTaskUseCase(repository, notifier)
complete_task_use_case = CompleteTaskUseCase(repository)


class CreateTaskRequest(BaseModel):
    title: str
    description: str
    owner_id: int
    owner_email: str


@app.post("/tasks")
def create_task(request: CreateTaskRequest):
    try:
        task = create_task_use_case.execute(
            request.title, request.description,
            request.owner_id, request.owner_email
        )
        return {"id": task.id, "title": task.title, "status": task.status.value}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@app.put("/tasks/{task_id}/complete")
def complete_task(task_id: int, user_id: int):
    try:
        task = complete_task_use_case.execute(task_id, user_id)
        return {"id": task.id, "status": task.status.value}
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))
    except PermissionError as e:
        raise HTTPException(status_code=403, detail=str(e))

מבנה הקבצים

taskflow/
    entities/
        task.py
        user.py
        project.py
    use_cases/
        task_use_cases.py
        user_use_cases.py
    adapters/
        sqlite_task_repository.py
        email_notifier.py
        sms_notifier.py
    main.py

מה מרוויחים?

החלפת מסד נתונים: רוצים לעבור מ-SQLite ל-PostgreSQL? כותבים PostgresTaskRepository שיורש מ-TaskRepository. אין שינוי בשכבת הישויות ומקרי השימוש.

בדיקות פשוטות: בדיקות ל-use cases לא צריכות מסד נתונים אמיתי:

def test_cannot_complete_task_of_another_user():
    repo = InMemoryTaskRepository()  # מסד נתונים מזויף לבדיקות
    use_case = CompleteTaskUseCase(repo)

    task = repo.save(Task(id=0, title="בדיקה", ..., owner_id=1))

    with pytest.raises(PermissionError):
        use_case.execute(task.id, requesting_user_id=2)  # משתמש אחר

הבדיקה רצה בלי שרת, בלי מסד נתונים אמיתי, בלי HTTP.