in-c4/main.py

52 lines
1.4 KiB
Python
Raw Permalink Normal View History

2022-08-04 05:41:51 +02:00
from fastapi import Depends, FastAPI, Response
from fastapi.staticfiles import StaticFiles
from sqlalchemy import event
from sqlalchemy.orm import Session
from sqlalchemy.engine import Engine
2022-08-06 02:38:47 +02:00
import re
2022-08-04 05:41:51 +02:00
import crud
from database import SessionLocal, engine
from models import Base
from schemas import Item
Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@event.listens_for(Engine, "connect")
def _set_sqlite_pragma(conn, _):
cursor = conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON;")
cursor.close()
@app.get("/api/items", response_model=dict[str, Item])
async def list_items(db: Session = Depends(get_db)):
2022-08-11 03:40:26 +02:00
# natural sort by id
natsort = lambda item: [int(t) if t.isdigit() else t.lower() for t in re.split('(\d+)', item.id)]
2022-08-04 05:41:51 +02:00
items = crud.get_items(db)
2022-08-06 02:38:47 +02:00
items = sorted(items, key=natsort)
2022-08-04 05:41:51 +02:00
return {i.id:i for i in items}
@app.put("/api/items/{id}")
async def put_item(id: str, item: Item, db: Session = Depends(get_db)):
if crud.put_item(db, id, item):
return Response(b'', status_code=204)
return Response(b'', status_code=201)
@app.delete("/api/items/{id}", status_code=204)
async def delete_item(id: str, db: Session = Depends(get_db)):
crud.delete_item(db, id)
app.mount("/", StaticFiles(directory="static", html=True), name="static")