After module 6: Using a Database

carsharing.py

@@ -1,77 +1,90 @@
 import uvicorn
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Depends
+from sqlmodel import create_engine, SQLModel, Session, select
 
-from schemas import load_db, save_db, CarInput, CarOutput, TripOutput, TripInput
+from schemas import CarInput, CarOutput, TripOutput, TripInput, Car, Trip
 
 app = FastAPI(title="Car Sharing")
-db = load_db()
+
+engine = create_engine(
+    "sqlite:///carsharing.db",
+    connect_args={"check_same_thread": False},  # Needed for SQLite
+    echo=True  # Log generated SQL
+)
+
+@app.on_event("startup")
+def on_startup():
+    SQLModel.metadata.create_all(engine)
+
+
+def get_session():
+    with Session(engine) as session:
+        yield session
 
 
 @app.get("/api/cars")
-def get_cars(size: str | None = None, doors: int | None = None) -> list:
-    result = db
+def get_cars(size: str | None = None, doors: int | None = None,
+             session: Session = Depends(get_session)) -> list:
+    query = select(Car)
     if size:
-        result = [car for car in result if car.size == size]
+        query = query.where(Car.size == size)
     if doors:
-        result = [car for car in result if car.doors >= doors]
-    return result
+        query = query.where(Car.doors >= doors)
+    return session.exec(query).all()
 
 
-@app.get("/api/cars/{id}")
-def car_by_id(id: int) -> dict:
-    result = [car for car in db if car.id == id]
-    if result:
-        return result[0]
+@app.get("/api/cars/{id}", response_model=CarOutput)
+def car_by_id(id: int, session: Session = Depends(get_session)) -> Car:
+    car = session.get(Car, id)
+    if car:
+        return car
     else:
         raise HTTPException(status_code=404, detail=f"No car with id={id}.")
 
 
-@app.post("/api/cars/", response_model=CarOutput)
-def add_car(car: CarInput) -> CarOutput:
-    new_car = CarOutput(size=car.size, doors=car.doors,
-                        fuel=car.fuel, transmission=car.transmission,
-                        id=len(db)+1)
-    db.append(new_car)
-    save_db(db)
+@app.post("/api/cars/", response_model=Car)
+def add_car(car_input: CarInput, session: Session = Depends(get_session)) -> Car:
+    new_car = Car.from_orm(car_input)
+    session.add(new_car)
+    session.commit()
+    session.refresh(new_car)
     return new_car
 
 
 @app.delete("/api/cars/{id}", status_code=204)
-def remove_car(id: int) -> None:
-    matches = [car for car in db if car.id == id]
-    if matches:
-        car = matches[0]
-        db.remove(car)
-        save_db(db)
+def remove_car(id: int, session: Session = Depends(get_session)) -> None:
+    car = session.get(Car, id)
+    if car:
+        session.delete(car)
+        session.commit()
     else:
         raise HTTPException(status_code=404, detail=f"No car with id={id}.")
 
 
-@app.put("/api/cars/{id}", response_model=CarOutput)
-def change_car(id: int, new_data: CarInput) -> CarOutput:
-    matches = [car for car in db if car.id == id]
-    if matches:
-        car = matches[0]
+@app.put("/api/cars/{id}", response_model=Car)
+def change_car(id: int, new_data: CarInput,
+               session: Session = Depends(get_session)) -> Car:
+    car = session.get(Car, id)
+    if car:
         car.fuel = new_data.fuel
         car.transmission = new_data.transmission
         car.size = new_data.size
         car.doors = new_data.doors
-        save_db(db)
+        session.commit()
         return car
     else:
         raise HTTPException(status_code=404, detail=f"No car with id={id}.")
 
 
-@app.post("/api/cars/{car_id}/trips", response_model=TripOutput)
-def add_trip(car_id: int, trip: TripInput) -> TripOutput:
-    matches = [car for car in db if car.id == car_id]
-    if matches:
-        car = matches[0]
-        new_trip = TripOutput(id=len(car.trips)+1,
-                              start=trip.start, end=trip.end,
-                              description=trip.description)
+@app.post("/api/cars/{car_id}/trips", response_model=Trip)
+def add_trip(car_id: int, trip_input: TripInput,
+             session: Session = Depends(get_session)) -> Trip:
+    car = session.get(Car, car_id)
+    if car:
+        new_trip = Trip.from_orm(trip_input, update={'car_id': car_id})
         car.trips.append(new_trip)
-        save_db(db)
+        session.commit()
+        session.refresh(new_trip)
         return new_trip
     else:
         raise HTTPException(status_code=404, detail=f"No car with id={id}.")

schemas.py

@@ -1,9 +1,7 @@
-import json
+from sqlmodel import SQLModel, Field, Relationship
 
-from pydantic import BaseModel
 
-
-class TripInput(BaseModel):
+class TripInput(SQLModel):
     start: int
     end: int
     description: str
@@ -13,7 +11,13 @@ class TripOutput(TripInput):
     id: int
 
 
-class CarInput(BaseModel):
+class Trip(TripInput, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    car_id: int = Field(foreign_key="car.id")
+    car: "Car" = Relationship(back_populates="trips")
+
+
+class CarInput(SQLModel):
     size: str
     fuel: str | None = "electric"
     doors: int
@@ -30,17 +34,12 @@ class Config:
         }
 
 
+class Car(CarInput, table=True):
+    id: int | None = Field(primary_key=True, default=None)
+    trips: list[Trip] = Relationship(back_populates="car")
+
+
 class CarOutput(CarInput):
     id: int
     trips: list[TripOutput] = []
 
-
-def load_db() -> list[CarOutput]:
-    """Load a list of Car objects from a JSON file"""
-    with open("cars.json") as f:
-        return [CarOutput.parse_obj(obj) for obj in json.load(f)]
-
-
-def save_db(cars: list[CarOutput]):
-    with open("cars.json", 'w') as f:
-        json.dump([car.dict() for car in cars], f, indent=4)