1.1 fastapi + postman פתרון

פתרון תרגול FastAPI – מערכת טיולים (Trips)

בפתרון זה נבנה API פשוט לניהול מערכת טיולים עם ישויות: טיול (Trip), מלון (Hotel), טיסה (Flight) ואטרקציה (Attraction).
כל הנתונים נשמרים ב־SQLite באמצעות SQLAlchemy. לכל ישות נגדיר פעולות בסיסיות לקריאה ויצירה:
- לקרוא את כל הרשומות ואת רשומה לפי מזהה (GET /{entity}, GET /{entity}/{id})
- ליצור רשומה חדשה (POST /{entity})
טיול כולל קישור לרשימות של מזהי מלונות, טיסות ואטרקציות.

---

התקנות

התקינו את התלויות הבאות לסביבה שלכם (ב venv):

pip install fastapi uvicorn sqlalchemy pydantic

---

מבנה פרויקט מומלץ

app/
  __init__.py        # קובץ ריק – מגדיר את התיקיה כ"חבילה" (package)
  database.py        # חיבור ל-SQLite והגדרת ה-Session
  models.py          # מודלי SQLAlchemy כולל טבלאות קישור
  controllers.py     # כל הcontroller-ים של הפרויקט
  schemas.py         # מודלי Pydantic ל-Request/Response
  main.py            # נקודות קצה FastAPI

הרצה:

uvicorn app.main:app --reload

---

קוד הפתרון

app/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./trips.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models.py

from sqlalchemy import Column, Integer, String, Table, ForeignKey
from sqlalchemy.orm import relationship
from .database import Base

# טבלאות קישור Many-to-Many בין טיולים לישויות אחרות
trip_hotels = Table(
    "trip_hotels",
    Base.metadata,
    Column("trip_id", ForeignKey("trips.id", ondelete="CASCADE"), primary_key=True),
    Column("hotel_id", ForeignKey("hotels.id", ondelete="CASCADE"), primary_key=True),
)

trip_flights = Table(
    "trip_flights",
    Base.metadata,
    Column("trip_id", ForeignKey("trips.id", ondelete="CASCADE"), primary_key=True),
    Column("flight_id", ForeignKey("flights.id", ondelete="CASCADE"), primary_key=True),
)

trip_attractions = Table(
    "trip_attractions",
    Base.metadata,
    Column("trip_id", ForeignKey("trips.id", ondelete="CASCADE"), primary_key=True),
    Column("attraction_id", ForeignKey("attractions.id", ondelete="CASCADE"), primary_key=True),
)


class Hotel(Base):
    __tablename__ = "hotels"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    city = Column(String, nullable=False)


class Flight(Base):
    __tablename__ = "flights"

    id = Column(Integer, primary_key=True, index=True)
    airline = Column(String, nullable=False)
    from_city = Column(String, nullable=False)
    to_city = Column(String, nullable=False)


class Attraction(Base):
    __tablename__ = "attractions"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    city = Column(String, nullable=False)


class Trip(Base):
    __tablename__ = "trips"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    description = Column(String, nullable=True)

    hotels = relationship("Hotel", secondary=trip_hotels, lazy="select")
    flights = relationship("Flight", secondary=trip_flights, lazy="select")
    attractions = relationship("Attraction", secondary=trip_attractions, lazy="select")

app/schemas.py

from typing import List, Optional
from pydantic import BaseModel, ConfigDict

# Hotels
class HotelBase(BaseModel):
    name: str
    city: str

class HotelCreate(HotelBase):
    pass

class HotelRead(HotelBase):
    id: int
    model_config = ConfigDict(from_attributes=True)

# Flights
class FlightBase(BaseModel):
    airline: str
    from_city: str
    to_city: str

class FlightCreate(FlightBase):
    pass

class FlightRead(FlightBase):
    id: int
    model_config = ConfigDict(from_attributes=True)

# Attractions
class AttractionBase(BaseModel):
    name: str
    city: str

class AttractionCreate(AttractionBase):
    pass

class AttractionRead(AttractionBase):
    id: int
    model_config = ConfigDict(from_attributes=True)

# Trips
class TripBase(BaseModel):
    name: str
    description: Optional[str] = None

class TripCreate(TripBase):
    hotel_ids: List[int] = []
    flight_ids: List[int] = []
    attraction_ids: List[int] = []

class TripRead(TripBase):
    id: int
    hotels: List[HotelRead] = []
    flights: List[FlightRead] = []
    attractions: List[AttractionRead] = []
    model_config = ConfigDict(from_attributes=True)

app/controllers.py

from typing import List
from fastapi import HTTPException, status
from sqlalchemy.orm import Session

from . import models, schemas

# ---------------------- Hotels ----------------------
def list_hotels(db: Session):
    return db.query(models.Hotel).all()


def get_hotel_by_id(hotel_id: int, db: Session):
    hotel = db.query(models.Hotel).get(hotel_id)
    if not hotel:
        raise HTTPException(status_code=404, detail="Hotel not found")
    return hotel


def create_hotel(hotel_in: schemas.HotelCreate, db: Session):
    hotel = models.Hotel(**hotel_in.dict())
    db.add(hotel)
    db.commit()
    db.refresh(hotel)
    return hotel


# ---------------------- Flights ----------------------
def list_flights(db: Session):
    return db.query(models.Flight).all()


def get_flight_by_id(flight_id: int, db: Session):
    flight = db.query(models.Flight).get(flight_id)
    if not flight:
        raise HTTPException(status_code=404, detail="Flight not found")
    return flight


def create_flight(flight_in: schemas.FlightCreate, db: Session):
    flight = models.Flight(**flight_in.dict())
    db.add(flight)
    db.commit()
    db.refresh(flight)
    return flight


# ---------------------- Attractions ----------------------
def list_attractions(db: Session):
    return db.query(models.Attraction).all()


def get_attraction_by_id(attraction_id: int, db: Session):
    attraction = db.query(models.Attraction).get(attraction_id)
    if not attraction:
        raise HTTPException(status_code=404, detail="Attraction not found")
    return attraction


def create_attraction(attraction_in: schemas.AttractionCreate, db: Session):
    attraction = models.Attraction(**attraction_in.dict())
    db.add(attraction)
    db.commit()
    db.refresh(attraction)
    return attraction


# ---------------------- Trips ----------------------
def list_trips(db: Session):
    return db.query(models.Trip).all()


def get_trip_by_id(trip_id: int, db: Session):
    trip = db.query(models.Trip).get(trip_id)
    if not trip:
        raise HTTPException(status_code=404, detail="Trip not found")
    return trip


def create_trip(trip_in: schemas.TripCreate, db: Session):
    hotels = []
    if trip_in.hotel_ids:
        hotels = db.query(models.Hotel).filter(models.Hotel.id.in_(trip_in.hotel_ids)).all()
        missing = set(trip_in.hotel_ids) - {h.id for h in hotels}
        if missing:
            raise HTTPException(status_code=400, detail=f"Missing hotels: {sorted(missing)}")

    flights = []
    if trip_in.flight_ids:
        flights = db.query(models.Flight).filter(models.Flight.id.in_(trip_in.flight_ids)).all()
        missing = set(trip_in.flight_ids) - {f.id for f in flights}
        if missing:
            raise HTTPException(status_code=400, detail=f"Missing flights: {sorted(missing)}")

    attractions = []
    if trip_in.attraction_ids:
        attractions = db.query(models.Attraction).filter(models.Attraction.id.in_(trip_in.attraction_ids)).all()
        missing = set(trip_in.attraction_ids) - {a.id for a in attractions}
        if missing:
            raise HTTPException(status_code=400, detail=f"Missing attractions: {sorted(missing)}")

    trip = models.Trip(name=trip_in.name, description=trip_in.description)
    trip.hotels = hotels
    trip.flights = flights
    trip.attractions = attractions

    db.add(trip)
    db.commit()
    db.refresh(trip)
    return trip

app/main.py

from typing import List
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session

from .database import Base, engine, get_db
from . import models, schemas
from . import controllers

app = FastAPI(title="Trips API")

Base.metadata.create_all(bind=engine)

# ---------------------- Hotels ----------------------
@app.get("/hotels", response_model=List[schemas.HotelRead])
def list_hotels_route(db: Session = Depends(get_db)):
    return controllers.list_hotels(db)

@app.get("/hotels/{hotel_id}", response_model=schemas.HotelRead)
def get_hotel_route(hotel_id: int, db: Session = Depends(get_db)):
    return controllers.get_hotel_by_id(hotel_id, db)

@app.post("/hotels", response_model=schemas.HotelRead, status_code=status.HTTP_201_CREATED)
def create_hotel_route(hotel_in: schemas.HotelCreate, db: Session = Depends(get_db)):
    return controllers.create_hotel(hotel_in, db)

# ---------------------- Flights ----------------------
@app.get("/flights", response_model=List[schemas.FlightRead])
def list_flights_route(db: Session = Depends(get_db)):
    return controllers.list_flights(db)

@app.get("/flights/{flight_id}", response_model=schemas.FlightRead)
def get_flight_route(flight_id: int, db: Session = Depends(get_db)):
    return controllers.get_flight_by_id(flight_id, db)

@app.post("/flights", response_model=schemas.FlightRead, status_code=status.HTTP_201_CREATED)
def create_flight_route(flight_in: schemas.FlightCreate, db: Session = Depends(get_db)):
    return controllers.create_flight(flight_in, db)

# ---------------------- Attractions ----------------------
@app.get("/attractions", response_model=List[schemas.AttractionRead])
def list_attractions_route(db: Session = Depends(get_db)):
    return controllers.list_attractions(db)

@app.get("/attractions/{attraction_id}", response_model=schemas.AttractionRead)
def get_attraction_route(attraction_id: int, db: Session = Depends(get_db)):
    return controllers.get_attraction_by_id(attraction_id, db)

@app.post("/attractions", response_model=schemas.AttractionRead, status_code=status.HTTP_201_CREATED)
def create_attraction_route(attraction_in: schemas.AttractionCreate, db: Session = Depends(get_db)):
    return controllers.create_attraction(attraction_in, db)

# ---------------------- Trips ----------------------
@app.get("/trips", response_model=List[schemas.TripRead])
def list_trips_route(db: Session = Depends(get_db)):
    return controllers.list_trips(db)

@app.get("/trips/{trip_id}", response_model=schemas.TripRead)
def get_trip_route(trip_id: int, db: Session = Depends(get_db)):
    return controllers.get_trip_by_id(trip_id, db)

@app.post("/trips", response_model=schemas.TripRead, status_code=status.HTTP_201_CREATED)
def create_trip_route(trip_in: schemas.TripCreate, db: Session = Depends(get_db)):
    return controllers.create_trip(trip_in, db)

---

בדיקה מהירה עם curl / Postman

לאחר הרצת השרת (uvicorn app.main:app --reload) ניתן לבדוק:

# יצירת מלון
curl -X POST http://127.0.0.1:8000/hotels \
  -H "Content-Type: application/json" \
  -d '{"name":"Hilton","city":"London"}'

# יצירת טיסה
curl -X POST http://127.0.0.1:8000/flights \
  -H "Content-Type: application/json" \
  -d '{"airline":"EL AL","from_city":"TLV","to_city":"LHR"}'

# יצירת אטרקציה
curl -X POST http://127.0.0.1:8000/attractions \
  -H "Content-Type: application/json" \
  -d '{"name":"London Eye","city":"London"}'

# שליפת כל הישויות
curl http://127.0.0.1:8000/hotels
curl http://127.0.0.1:8000/flights
curl http://127.0.0.1:8000/attractions

# יצירת טיול עם קישורים לישויות שנוצרו (נניח שהתקבלו id=1 לכל אחת)
curl -X POST http://127.0.0.1:8000/trips \
  -H "Content-Type: application/json" \
  -d '{"name":"London Trip","description":"A short trip","hotel_ids":[1],"flight_ids":[1],"attraction_ids":[1]}'

# שליפה
curl http://127.0.0.1:8000/trips
curl http://127.0.0.1:8000/trips/1

ניתן גם להיעזר ב־Swagger UI: גלשו ל־http://127.0.0.1:8000/docs ובצעו קריאות דרך הדפדפן.