66 lines
1.7 KiB
Python
66 lines
1.7 KiB
Python
import re
|
|
import time
|
|
|
|
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("(\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}")
|
|
async def put_item(id: str, item: Item, db: Session = Depends(get_db)):
|
|
if item.last_updated is None:
|
|
item.last_updated = rough_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 rough_timestamp() -> int:
|
|
"""Provides an current timestamp with reduced resolution, to improve anonymity."""
|
|
granularity = 2**20 # about 12 days
|
|
return int(time.time()) // granularity * granularity
|
|
|
|
|
|
app.mount("/", StaticFiles(directory="static", html=True), name="static")
|