3.2 ארכיטקטורת שכבות הרצאה

ארכיטקטורת שכבות - Layered Architecture

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

---

שלוש השכבות הקלאסיות

+---------------------------+
|   Presentation Layer      |  שכבת ממשק המשתמש (Routes, HTTP)
+---------------------------+
           |
           v
+---------------------------+
|   Business Logic Layer    |  שכבת הלוגיקה העסקית (Services, Use Cases)
+---------------------------+
           |
           v
+---------------------------+
|   Data Access Layer       |  שכבת גישה לנתונים (Repositories, DB)
+---------------------------+

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

---

שכבה 1: Presentation Layer - שכבת ממשק

שכבה זו מטפלת בכל דבר הקשור ל-HTTP:
- מקבלת בקשות HTTP
- מבצעת וולידציה בסיסית של הקלט
- מעבירה לשכבת הלוגיקה
- מחזירה תשובת HTTP

# routes/tasks.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from services.task_service import TaskService

router = APIRouter(prefix="/tasks")
task_service = TaskService()


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


class TaskResponse(BaseModel):
    id: int
    title: str
    status: str


@router.post("/", response_model=TaskResponse, status_code=201)
def create_task(request: CreateTaskRequest):
    try:
        task = task_service.create_task(
            request.title,
            request.description,
            request.owner_id
        )
        return TaskResponse(id=task["id"], title=task["title"], status=task["status"])
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.get("/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    task = task_service.get_task(task_id)
    if not task:
        raise HTTPException(status_code=404, detail="משימה לא נמצאה")
    return TaskResponse(**task)


@router.put("/{task_id}/complete")
def complete_task(task_id: int, user_id: int):
    try:
        task_service.complete_task(task_id, user_id)
        return {"message": "משימה הושלמה"}
    except PermissionError as e:
        raise HTTPException(status_code=403, detail=str(e))

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

---

שכבה 2: Business Logic Layer - שכבת הלוגיקה

שכבה זו היא "המוח" של האפליקציה:
- מממשת את כללי הביזנס
- מתאמת בין repositories שונים
- שולחת notifications
- לא יודעת שום דבר על HTTP

# services/task_service.py
from repositories.task_repository import TaskRepository
from repositories.user_repository import UserRepository
from services.notification_service import NotificationService


class TaskService:
    def __init__(self):
        self._task_repo = TaskRepository()
        self._user_repo = UserRepository()
        self._notifier = NotificationService()

    def create_task(self, title: str, description: str, owner_id: int) -> dict:
        # לוגיקה עסקית: וולידציה
        if len(title.strip()) < 3:
            raise ValueError("כותרת המשימה חייבת להכיל לפחות 3 תווים")

        # לוגיקה עסקית: בדיקת הרשאות
        user = self._user_repo.find_by_id(owner_id)
        if not user:
            raise ValueError(f"משתמש {owner_id} לא נמצא")

        # יצירת המשימה
        task = self._task_repo.save({
            "title": title.strip(),
            "description": description,
            "status": "todo",
            "owner_id": owner_id
        })

        # שליחת התראה
        self._notifier.notify_task_created(task, user["email"])

        return task

    def get_task(self, task_id: int) -> dict | None:
        return self._task_repo.find_by_id(task_id)

    def complete_task(self, task_id: int, requesting_user_id: int):
        task = self._task_repo.find_by_id(task_id)
        if not task:
            raise ValueError(f"משימה {task_id} לא נמצאה")

        # לוגיקה עסקית: הרשאה
        if task["owner_id"] != requesting_user_id:
            raise PermissionError("רק הבעלים יכול להשלים את המשימה")

        self._task_repo.update(task_id, status="done")

---

שכבה 3: Data Access Layer - שכבת הנתונים

שכבה זו מדברת עם מסד הנתונים בלבד:
- שמירה, שליפה, עדכון, מחיקה
- ממירה בין פורמט DB לפורמט Python
- לא יודעת על לוגיקה עסקית

# repositories/task_repository.py
import sqlite3
from datetime import datetime


class TaskRepository:
    def __init__(self, db_path: str = "taskflow.db"):
        self._db_path = db_path

    def _get_db(self):
        return sqlite3.connect(self._db_path)

    def save(self, task: dict) -> dict:
        db = self._get_db()
        cursor = db.execute(
            "INSERT INTO tasks (title, description, status, owner_id, created_at) VALUES (?, ?, ?, ?, ?)",
            (task["title"], task["description"], task["status"],
             task["owner_id"], datetime.now().isoformat())
        )
        db.commit()
        return {**task, "id": cursor.lastrowid}

    def find_by_id(self, task_id: int) -> dict | None:
        db = self._get_db()
        row = db.execute(
            "SELECT id, title, description, status, owner_id FROM tasks WHERE id=?",
            (task_id,)
        ).fetchone()
        if not row:
            return None
        return {
            "id": row[0], "title": row[1], "description": row[2],
            "status": row[3], "owner_id": row[4]
        }

    def find_by_owner(self, owner_id: int) -> list[dict]:
        db = self._get_db()
        rows = db.execute(
            "SELECT id, title, description, status, owner_id FROM tasks WHERE owner_id=?",
            (owner_id,)
        ).fetchall()
        return [
            {"id": r[0], "title": r[1], "description": r[2], "status": r[3], "owner_id": r[4]}
            for r in rows
        ]

    def update(self, task_id: int, **fields):
        if not fields:
            return
        set_clause = ", ".join(f"{k}=?" for k in fields)
        values = list(fields.values()) + [task_id]
        db = self._get_db()
        db.execute(f"UPDATE tasks SET {set_clause} WHERE id=?", values)
        db.commit()

---

מבנה הקבצים

taskflow/
    main.py
    routes/
        tasks.py
        users.py
        projects.py
    services/
        task_service.py
        user_service.py
        notification_service.py
    repositories/
        task_repository.py
        user_repository.py
        project_repository.py
    models/
        task.py          (Pydantic schemas)
        user.py

היתרונות של ארכיטקטורת שכבות

testability: אפשר לבדוק כל שכבה בנפרד. service tests עם mock repositories.

maintainability: רוצים לעבור מ-SQLite ל-PostgreSQL? משנים רק את שכבת ה-repositories.

readability: מפתח חדש יודע לאן ללכת - לוגיקה ב-service, DB ב-repository.

separation of concerns: כל שכבה יודעת רק מה שהיא צריכה לדעת.

מלכודת נפוצה: Anemic Domain Model

שכבות לא אומרות שה-entity הוא רק מבנה נתונים ריק. לוגיקה שמתייחסת לentity צריכה להיות בו:

# לא טוב - Task ריק מלוגיקה
class Task:
    def __init__(self, id, title, status): ...

class TaskService:
    def is_overdue(self, task): ...  # לוגיקה של Task נמצאת בservice?


# טוב - Task מכיל לוגיקה שקשורה לו
class Task:
    def is_overdue(self) -> bool:
        return self.due_date and datetime.now() > self.due_date and self.status != "done"