128 lines
3.3 KiB
Python
128 lines
3.3 KiB
Python
import datetime
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.encoders import jsonable_encoder
|
|
from sqlalchemy.exc import IntegrityError, DataError
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
from contextlib import asynccontextmanager
|
|
from api.models import Car, FuelItem
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
from fastapi.responses import JSONResponse
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
database_url: str = ""
|
|
sqlite_file_name: str = ""
|
|
|
|
model_config = SettingsConfigDict(env_file=".env")
|
|
|
|
|
|
settings = Settings()
|
|
|
|
database_url = settings.database_url
|
|
|
|
connect_args = {
|
|
"check_same_thread": False
|
|
}
|
|
engine = create_engine(database_url, echo=True, connect_args=connect_args)
|
|
|
|
|
|
def create_db_and_tables():
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
|
|
def fill_database():
|
|
car = Car(id_string="Auto 1")
|
|
with Session(engine) as session:
|
|
session.add(car)
|
|
session.commit()
|
|
session.refresh(car)
|
|
fill_item = FuelItem(
|
|
car_id=car.id,
|
|
date=datetime.date.today(),
|
|
odometer_reading=101005,
|
|
fuel_fill=52.26
|
|
)
|
|
session.add(fill_item)
|
|
session.commit()
|
|
session.refresh(fill_item)
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
@asynccontextmanager
|
|
async def lifespan(actual_app: FastAPI):
|
|
# startup
|
|
create_db_and_tables()
|
|
try:
|
|
fill_database()
|
|
except IntegrityError as e:
|
|
print(e)
|
|
yield
|
|
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
|
|
def save_object(obj):
|
|
with Session(engine) as session:
|
|
try:
|
|
session.add(obj)
|
|
session.commit()
|
|
session.refresh(obj)
|
|
return obj
|
|
except IntegrityError|DataError as e:
|
|
content = jsonable_encoder({"message": e})
|
|
return JSONResponse(content=content, status_code=400)
|
|
|
|
|
|
def update_object(obj_id: int, obj):
|
|
with Session(engine) as session:
|
|
try:
|
|
object_db = session.get(type(obj), obj_id)
|
|
if not object_db:
|
|
content = jsonable_encoder({"message": "%s-ID not found" % type(obj).__qualname__})
|
|
return JSONResponse(content=content, status_code=404)
|
|
obj_data = obj.model_dump(exclude_unset=True)
|
|
object_db.sqlmodel_update(obj_data)
|
|
session.add(object_db)
|
|
session.commit()
|
|
session.refresh(object_db)
|
|
return object_db
|
|
except IntegrityError|DataError as e:
|
|
content = jsonable_encoder({"message": e})
|
|
return JSONResponse(content=content, status_code=400)
|
|
|
|
|
|
@app.post("/cars/")
|
|
def create_cars(car: Car):
|
|
return save_object(car)
|
|
|
|
|
|
@app.put("/cars/{obj_id}")
|
|
def update_cars(obj_id, car: Car):
|
|
return update_object(obj_id, car)
|
|
|
|
|
|
@app.get("/cars/")
|
|
def read_cars():
|
|
with Session(engine) as session:
|
|
cars = session.exec(select(Car)).all()
|
|
return cars
|
|
|
|
|
|
@app.post("/fill_items/")
|
|
def create_cars(fill_item: FuelItem):
|
|
return save_object(fill_item)
|
|
|
|
|
|
@app.put("/fill_items/{obj_id}")
|
|
def update_cars(obj_id, fill_item: FuelItem):
|
|
return update_object(obj_id, fill_item)
|
|
|
|
|
|
@app.get("/fill_items/")
|
|
def read_cars():
|
|
with Session(engine) as session:
|
|
fill_items = session.exec(select(FuelItem)).all()
|
|
return fill_items
|