in-c4/main.py
2024-02-01 01:04:49 +01:00

70 lines
1.8 KiB
Python

import re
from calendar import timegm
from datetime import date
from fastapi import Depends, FastAPI, Response
from fastapi.staticfiles import StaticFiles
from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session
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)):
# natural sort by id
natsort = lambda item: [
int(t) if t.isdigit() else t.lower() for t in re.split(r"(\d+)", item.id)
]
items = crud.get_items(db)
items = sorted(items, key=natsort)
return {i.id: i for i in items}
@app.put(
"/api/items/{id}",
status_code=201,
responses={201: {"description": "created"}, 204: {"description": "updated"}},
)
async def put_item(id: str, item: Item, db: Session = Depends(get_db)):
if item.last_updated is None:
item.last_updated = month_timestamp()
if crud.put_item(db, id, item) == crud.PutItemResult.UPDATED:
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)
def month_timestamp() -> int:
"""Provides the timestamp of the current month's beginning (for improved privacy)"""
return timegm(date.today().replace(day=1).timetuple())
app.mount("/", StaticFiles(directory="static", html=True), name="static")