בשלב הבא אפשר להחליף את dict במסד נתונים (PostgreSQL או MongoDB), להוסיף JWT/OAuth, ולהפוך endpoints לאסינכרוניים (async def) כשיש IO כבד.
הנקודה היא: אל תחליף הכל יחד. שומרים את החוזה, משפרים שכבה אחת בכל פעם.
CRUD, סכמות Pydantic, סטטוסים, ושגיאות אחידות
המטרה היא להראות API קטן אך "נכון": סכמות עם Pydantic, CRUD, סטטוסים, ושגיאות עקביות. זה הבסיס למערכת שמתחברת לסוכנים ולפרודקשן.
ב-REST, ה-API הוא חוזה. FastAPI + Pydantic מאפשרים להגדיר חוזה ולבצע ולידציה אוטומטית. זה מאפשר פיתוח מקבילי של לקוח ושרת, ומקטין באגים.
from pydantic import BaseModel, Field
from datetime import datetime
class NoteCreate(BaseModel):
title: str = Field(min_length=1, max_length=80)
content: str = Field(default="", max_length=10_000)
class Note(NoteCreate):
id: int
created_at: datetime
כדי להתמקד ב-API, נתחיל במאגר בזיכרון (dict). בפרודקשן מחליפים למסד נתונים, אבל החוזה וה-endpoints נשארים זהים.
from fastapi import FastAPI, HTTPException
from datetime import datetime
from models import Note, NoteCreate
app = FastAPI(title="Notes API", version="1.0")
notes: dict[int, Note] = {}
next_id = 1
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/api/v1/notes", response_model=Note, status_code=201)
def create_note(payload: NoteCreate):
global next_id
note = Note(id=next_id, created_at=datetime.utcnow(), **payload.model_dump())
notes[note.id] = note
next_id += 1
return note
@app.get("/api/v1/notes", response_model=list[Note])
def list_notes():
return list(notes.values())
@app.get("/api/v1/notes/{note_id}", response_model=Note)
def get_note(note_id: int):
note = notes.get(note_id)
if not note:
raise HTTPException(status_code=404, detail="Note not found")
return note
@app.put("/api/v1/notes/{note_id}", response_model=Note)
def replace_note(note_id: int, payload: NoteCreate):
if note_id not in notes:
raise HTTPException(status_code=404, detail="Note not found")
old = notes[note_id]
updated = Note(id=note_id, created_at=old.created_at, **payload.model_dump())
notes[note_id] = updated
return updated
@app.delete("/api/v1/notes/{note_id}", status_code=204)
def delete_note(note_id: int):
if note_id not in notes:
raise HTTPException(status_code=404, detail="Note not found")
del notes[note_id]
FastAPI מפרסם תיעוד ב-/docs ו-/redoc. זה הופך את החוזה לכלי עבודה אמיתי לצוותים.
במערכות מרובות שירותים - OpenAPI הוא השפה המשותפת.
Endpoint כמו /health מאפשר לניטור ול-load balancer לדעת אם השירות חי. במערכת מלאה מוסיפים בדיקות DB/Cache.
מומלץ גם readiness + liveness כשעובדים עם קונטיינרים.
בשלב הבא אפשר להחליף את dict במסד נתונים (PostgreSQL או MongoDB), להוסיף JWT/OAuth, ולהפוך endpoints לאסינכרוניים (async def) כשיש IO כבד.
הנקודה היא: אל תחליף הכל יחד. שומרים את החוזה, משפרים שכבה אחת בכל פעם.
כך צד לקוח מתקשר עם השרת. זה אותו רעיון גם כאשר "לקוח" הוא שירות אחר או סוכן.
async function createNote() {
const res = await fetch("/api/v1/notes", {
method: "POST",
headers: {"Content-Type": "application/json"},
body: JSON.stringify({title: "Hello", content: "From browser"})
});
if (!res.ok) throw new Error(await res.text());
return await res.json();
}
כעת יש לנו API אמיתי. סוכן יכול לקרוא לו ככלי (Tool) ולהרכיב תהליכים.