pyr0ball
a90c3fc172
Wraps the Task 2 CRUD layer in a FastAPI app (schemas.py + main.py) with POST/GET/GET-by-id/PATCH routes for items, matching the exact paths and status codes the Discord bot's HTTP client will depend on. Also fixes a latent bug in db.get_engine: sqlite:///:memory: without a StaticPool gives each new session a fresh, empty database, which broke as soon as more than one session shared an engine (the API's per-request session pattern). Tasks 1-2 never hit this because their tests used a single session per engine.
66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
import os
|
|
from fastapi import FastAPI, HTTPException, Response
|
|
from app import crud
|
|
from app.db import get_engine, make_session_factory, init_db
|
|
from app.schemas import ItemCreate, ItemUpdate, ItemOut
|
|
|
|
|
|
def create_app(session_factory=None) -> FastAPI:
|
|
if session_factory is None:
|
|
db_url = os.environ.get("CHORUS_DB_URL", "sqlite:///./chorus.db")
|
|
engine = get_engine(db_url)
|
|
init_db(engine)
|
|
session_factory = make_session_factory(engine)
|
|
|
|
app = FastAPI(title="Chorus API")
|
|
|
|
@app.post("/items", response_model=ItemOut)
|
|
def post_item(payload: ItemCreate, response: Response):
|
|
session = session_factory()
|
|
try:
|
|
item, created = crud.create_item(session, **payload.model_dump())
|
|
response.status_code = 201 if created else 200
|
|
return item
|
|
finally:
|
|
session.close()
|
|
|
|
@app.get("/items", response_model=list[ItemOut])
|
|
def get_items(stage: str | None = None, modality: str | None = None,
|
|
include_done: bool = False):
|
|
session = session_factory()
|
|
try:
|
|
return crud.list_items(session, stage=stage, modality=modality,
|
|
include_done=include_done)
|
|
finally:
|
|
session.close()
|
|
|
|
@app.get("/items/{item_id}", response_model=ItemOut)
|
|
def get_item(item_id: int):
|
|
session = session_factory()
|
|
try:
|
|
item = crud.get_item(session, item_id)
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
return item
|
|
finally:
|
|
session.close()
|
|
|
|
@app.patch("/items/{item_id}", response_model=ItemOut)
|
|
def patch_item(item_id: int, payload: ItemUpdate):
|
|
session = session_factory()
|
|
try:
|
|
fields = {k: v for k, v in payload.model_dump().items() if v is not None}
|
|
try:
|
|
item = crud.update_item(session, item_id, **fields)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc))
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
return item
|
|
finally:
|
|
session.close()
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|