"""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.available import router as available_router from app.catalog import store from app.db import db_healthy, init_db from app.inventory import router as inventory_router from app.orders import router as orders_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) app.include_router(available_router) app.include_router(orders_router) VALID_CATEGORIES = {"ships", "ship-items", "fps-items", "commodities"} 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, {})}