Files
fpov-exchange/trade-hub/backend/app/api.py

124 lines
3.9 KiB
Python

"""FastAPI routes for the Star Citizen inventory (scunpacked data).
The catalog endpoints below are read-only. Accounts and personal inventories
are served by the auth and inventory routers and persisted in PostgreSQL.
"""
import math
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from app import config
from app.auth import router as auth_router
from app.catalog import store
from app.db import db_healthy, init_db
from app.inventory import router as inventory_router
@asynccontextmanager
async def lifespan(_app: FastAPI):
if config.AUTO_INIT_DB:
init_db(wait_seconds=config.DB_WAIT_SECONDS)
yield
app = FastAPI(
title="Star Citizen Inventory API",
description="Read-only item catalog built from scunpacked game data, plus "
"per-user accounts and personal inventories.",
version="1.1.0",
lifespan=lifespan,
)
# Same-origin deployment by default. Cross-origin access must be opted into
# explicitly, because credentialed cookies and a wildcard origin are
# incompatible (and unsafe).
if config.CORS_ORIGINS:
app.add_middleware(
CORSMiddleware,
allow_origins=config.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["GET", "POST", "PATCH", "DELETE"],
allow_headers=["Content-Type"],
)
app.include_router(auth_router)
app.include_router(inventory_router)
VALID_CATEGORIES = {"ships", "ship-items", "fps-items"}
def _filter_items(category, search, manufacturer, item_type, size):
result = store.items
if category is not None:
if category not in VALID_CATEGORIES:
raise HTTPException(
status_code=422,
detail=f"Invalid category '{category}'. Valid: {sorted(VALID_CATEGORIES)}",
)
result = [i for i in result if i["category"] == category]
if search is not None and search.strip():
needle = search.strip().lower()
result = [
i
for i in result
if needle in i["name"].lower() or needle in i["class_name"].lower()
]
if manufacturer is not None and manufacturer.strip():
code = manufacturer.strip().lower()
result = [i for i in result if i["manufacturer"]["code"].lower() == code]
if item_type is not None and item_type.strip():
wanted = item_type.strip().lower()
result = [i for i in result if (i["type"] or "").lower() == wanted]
if size is not None:
result = [i for i in result if i["size"] == size]
return result
@app.get("/healthz")
def healthz():
return {"status": "ok", "counts": store.counts, "db": db_healthy()}
@app.get("/api/v1/stats")
def stats():
return {"categories": store.counts, "manufacturers": len(store.manufacturers)}
@app.get("/api/v1/manufacturers")
def manufacturers():
return store.manufacturers
@app.get("/api/v1/items")
def list_items(
category: str | None = None,
search: str | None = None,
manufacturer: str | None = None,
item_type: str | None = Query(default=None, alias="type"),
size: int | None = None,
page: int = Query(default=1, ge=1),
page_size: int = Query(default=25, ge=1),
):
page_size = min(page_size, 100) # cap, don't reject
filtered = _filter_items(category, search, manufacturer, item_type, size)
total = len(filtered)
total_pages = max(1, math.ceil(total / page_size))
start = (page - 1) * page_size
return {
"total": total,
"page": page,
"page_size": page_size,
"total_pages": total_pages,
"items": filtered[start : start + page_size],
}
@app.get("/api/v1/items/{class_name}")
def item_detail(class_name: str):
summary = store.by_class_name.get(class_name)
if summary is None:
raise HTTPException(status_code=404, detail=f"Item '{class_name}' not found")
return {**summary, "details": store.details.get(class_name, {})}