DB + Pydantic Validation

כשעוברים מפרויקט דמו לפרודקשן, הבעיה המרכזית היא יציבות נתונים: מה נכנס למערכת, איך מאחסנים, ואיך מוודאים שהחוזה לא נשבר. כאן נכנסים Pydantic, שכבת Persistence, ו-DB.

רעיון מפתח: ב-API מודרני לא "זורקים dict". מגדירים סכמות, מבדילים בין Create/Read/Update, ומחזירים שגיאות עקביות. ה-DB הוא מקור האמת - וה-API הוא שכבת האכיפה.

1) שכבות במערכת Backend טיפוסית

סטודנטים רבים כותבים הכל בתוך ה-endpoint. זה עובד לשעה הראשונה ואז מתחיל להישבר. במקום זה מפרידים אחריות:

חלוקה מינימלית שמאפשרת סקייל: routes מציגים API, סכמות מגדירות חוזה, שירותים מיישמים לוגיקה, ו-persistence מדבר עם DB.

2) למה Pydantic הוא הרבה יותר מ-"type hints"

Pydantic נותן לך שלושה דברים שמבדילים מערכת יציבה ממערכת שבירה:

ולידציה: בדיקה אוטומטית של קלט - אורכים, טווחים, פורמטים, וכפייה של טיפוסים.
סיריאליזציה: המרה עקבית ל-JSON, כולל תאריכים, enums, וחבילות נתונים מורכבות.
Contract: תיעוד אוטומטי ב-OpenAPI, שמאפשר ללקוח לייצר SDK ולצוותים לעבוד במקביל.

3) תבנית מודלים נכונה: Create, Read, Update

הטעות הקלאסית היא להשתמש באותו מודל לכל המצבים. בפועל אלו חוזים שונים:

Create

מה הלקוח רשאי לשלוח. בדרך כלל בלי id ובלי fields שמחושבים בשרת.

Read

מה השרת מחזיר. כולל id, timestamps, ולעתים שדות מחושבים.

Update

עדכון חלקי או מלא. לעדכון חלקי נדרש מודל עם optional fields.

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional

class NoteCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    content: str = Field(default="", max_length=10_000)

class NoteUpdate(BaseModel):
    title: Optional[str] = Field(default=None, min_length=1, max_length=80)
    content: Optional[str] = Field(default=None, max_length=10_000)

class NoteRead(BaseModel):
    id: int
    title: str
    content: str
    created_at: datetime

למה Update שונה מ-Create?

ב-PATCH הלקוח יכול לשלוח רק חלק מהשדות. אם תשתמש ב-Create, אתה תכריח אותו לשלוח הכל, או שתדרוס ערכים קיימים בטעות. לכן update model עם Optional הוא דפוס חשוב.

4) מעבר מ-dict ל-DB אמיתי

In-memory dict הוא כלי לימודי מעולה, אבל בפרודקשן צריך:

התמדה: נתונים שורדים ריסטארט.
קונקרנציה: מספר תהליכים/שרתים כותבים יחד בלי לדרוס.
שאילתות: פילטרים, מיון, אינדקסים.
אכיפת חוקים: unique constraints, foreign keys, transactions.

5) SQLModel/SQLAlchemy - דוגמת מינימום עובדת

הדוגמה הבאה משתמשת ב-SQLModel (שבנוי מעל SQLAlchemy) כדי להגדיר טבלת Notes ולבנות CRUD אמיתי עם SQLite. זה פתרון מצוין לסטודנטים כי הוא קריא, ומחבר DB + מודלים.

from sqlmodel import SQLModel, create_engine, Session

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, echo=False)

def init_db() -> None:
    SQLModel.metadata.create_all(engine)

def get_session() -> Session:
    return Session(engine)

from datetime import datetime
from sqlmodel import SQLModel, Field

class Note(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    content: str = ""
    created_at: datetime = Field(default_factory=datetime.utcnow)

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from db import init_db, get_session
from models import Note
from schemas import NoteCreate, NoteUpdate, NoteRead

app = FastAPI(title="Notes API (DB)", version="1.0")

@app.on_event("startup")
def on_startup():
    init_db()

def session_dep():
    with get_session() as session:
        yield session

@app.post("/api/v1/notes", response_model=NoteRead, status_code=201)
def create_note(payload: NoteCreate, session: Session = Depends(session_dep)):
    note = Note(title=payload.title, content=payload.content)
    session.add(note)
    session.commit()
    session.refresh(note)
    return NoteRead.model_validate(note.__dict__ | {"id": note.id, "created_at": note.created_at})

@app.get("/api/v1/notes", response_model=list[NoteRead])
def list_notes(session: Session = Depends(session_dep)):
    notes = session.exec(select(Note).order_by(Note.id.desc())).all()
    return [NoteRead.model_validate(n.__dict__ | {"id": n.id, "created_at": n.created_at}) for n in notes]

@app.get("/api/v1/notes/{note_id}", response_model=NoteRead)
def get_note(note_id: int, session: Session = Depends(session_dep)):
    note = session.get(Note, note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    return NoteRead.model_validate(note.__dict__ | {"id": note.id, "created_at": note.created_at})

@app.patch("/api/v1/notes/{note_id}", response_model=NoteRead)
def update_note(note_id: int, payload: NoteUpdate, session: Session = Depends(session_dep)):
    note = session.get(Note, note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")

    if payload.title is not None:
        note.title = payload.title
    if payload.content is not None:
        note.content = payload.content

    session.add(note)
    session.commit()
    session.refresh(note)
    return NoteRead.model_validate(note.__dict__ | {"id": note.id, "created_at": note.created_at})

@app.delete("/api/v1/notes/{note_id}", status_code=204)
def delete_note(note_id: int, session: Session = Depends(session_dep)):
    note = session.get(Note, note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    session.delete(note)
    session.commit()

נקודות פרודקשן מינימליות סביב DB

Migration: בפרודקשן לא יוצרים טבלאות ישירות מהקוד, אלא משתמשים ב-Alembic.
Connection pool: ל-Postgres משתמשים ב-pool, ובענן מכוונים max connections.
Transactions: פעולות מרובות כתיבות צריכות להיות אטומיות - commit אחד או rollback.
Indexes: שדות חיפוש ומיון צריכים אינדקס, אחרת latency קופץ.

6) שגיאות עקביות - Error Schema

FastAPI יחזיר אוטומטית 422 על ולידציה. אבל לשגיאות דומיין (למשל "כותרת כבר קיימת") מומלץ להגדיר פורמט שגיאה אחיד:

{
  "error": {
    "code": "NOTE_NOT_FOUND",
    "message": "No note with id=123",
    "trace_id": "a7f1c1..."
  }
}

היתרון: צד לקוח יודע לתרגם קודים, לוגים יכולים להיות מקושרים, וסוכן יכול להחליט איך לתקן (למשל לשאול את המשתמש שאלה נוספת).

תרגול

שאלות בדיקה (10)

למה Create/Read/Update הם חוזים שונים?
מה ההבדל בין PUT ל-PATCH מבחינת מודלים?
למה DB פותר בעיות ש-dict לא יכול לפתור?
מה role של SQLAlchemy/SQLModel במערכת?
מה ההבדל בין commit ל-refresh?
מתי נדרשת transaction אמיתית?
מה FastAPI מחזיר אוטומטית על ולידציה שנכשלה?
למה חשוב error schema אחיד?
למה migrations (Alembic) חשובים בפרודקשן?
אילו שדות היית מאנדקס במערכת Notes גדולה?

משימת בית (CRUD עם DB אמיתי)

בנה שירות Notes עם DB (SQLite מספיק) וארבעה endpoints:

POST   /api/v1/notes
GET    /api/v1/notes
PATCH  /api/v1/notes/{id}
DELETE /api/v1/notes/{id}

דרישות:

הפרדה ברורה בין SQLModel table לבין Pydantic schemas (Create/Read/Update).
ולידציה: title מינימום 1, מקסימום 80. content מקסימום 10,000.
החזרת סטטוסים נכונים: 201 ביצירה, 204 במחיקה, 404 כשאין משאב.
הוספת endpoint /health שמחזיר {"status":"ok"}.

בדיקת פתרון (Checklist + בדיקות)

Checklist:

עולה מקומית ויוצר app.db בזמן startup.
POST מחזיר 201 עם id ו-created_at.
PATCH מאפשר לעדכן שדה אחד בלי לדרוס את השני.
DELETE מחזיר 204, וקריאה אחרי מחיקה מחזירה 404.
ולידציה נכשלת מחזירה 422.

בדיקת smoke עם curl:

# create
curl -X POST http://localhost:8000/api/v1/notes \
  -H "Content-Type: application/json" \
  -d '{"title":"First","content":"Hello DB"}'

# list
curl http://localhost:8000/api/v1/notes

# patch
curl -X PATCH http://localhost:8000/api/v1/notes/1 \
  -H "Content-Type: application/json" \
  -d '{"content":"Updated"}'

# delete
curl -X DELETE -i http://localhost:8000/api/v1/notes/1

בדיקות יחידה (pytest) - כיוון:

def test_create_then_get(client):
    r = client.post("/api/v1/notes", json={"title":"t","content":"c"})
    assert r.status_code == 201
    note_id = r.json()["id"]

    r2 = client.get(f"/api/v1/notes/{note_id}")
    assert r2.status_code == 200
    assert r2.json()["title"] == "t"

המשך טבעי מכאן הוא לעטוף את השירות הזה ב-Docker ולהרים אותו ב-Cloud Run בצורה נכונה (env, secrets, health, scaling). כשתכתוב "המשך" אכין את הפרק Docker + Cloud Run.