93 lines
2.1 KiB
Python
93 lines
2.1 KiB
Python
import datetime
|
|
|
|
from fastapi import FastAPI
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
from contextlib import asynccontextmanager
|
|
from api.models import Car, FuelItem
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
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)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# startup
|
|
create_db_and_tables()
|
|
try:
|
|
fill_database()
|
|
except IntegrityError as e:
|
|
print(e)
|
|
yield
|
|
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
|
|
@app.post("/cars/")
|
|
def create_cars(car: Car):
|
|
with Session(engine) as session:
|
|
session.add(car)
|
|
session.commit()
|
|
session.refresh(car)
|
|
return 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):
|
|
with Session(engine) as session:
|
|
session.add(fill_item)
|
|
session.commit()
|
|
session.refresh(fill_item)
|
|
return fill_item
|
|
|
|
|
|
@app.get("/fill_items/")
|
|
def read_cars():
|
|
with Session(engine) as session:
|
|
fill_items = session.exec(select(FuelItem)).all()
|
|
return fill_items
|