Spaces:
Running
Running
| """ | |
| app.py | |
| FastAPI app: a small JSON API over server/deckbuilder.py, plus the built | |
| React frontend (web/dist) when it exists. | |
| Run from the repo root (the training code loads ./tokenizer.json and | |
| ./data/*.pt by relative path): | |
| uvicorn server.app:app --port 8000 | |
| Completions run one at a time (DeckBuilder holds a lock), so at most | |
| MTG_MAX_WAITING (default 4) may wait behind the running one; past that, | |
| /api/complete answers 503 at once instead of queueing without bound. | |
| MTG_API_DOCS=0 turns off FastAPI's /docs, /redoc and /openapi.json. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import threading | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| from typing import Literal | |
| from fastapi import FastAPI, Query, Request | |
| from fastapi.middleware.gzip import GZipMiddleware | |
| from fastapi.responses import FileResponse, JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from pydantic import BaseModel, Field | |
| from server.deckbuilder import DEFAULT_TARGET_LANDS, MIN_TARGET_LANDS, DeckBuilder, DeckInputError | |
| from server.card_prices import PriceBook, downloads_enabled | |
| from server.card_stats import ensure_card_stats | |
| from server.model_files import sync_model_files, update_check_enabled | |
| WEB_DIST = Path(__file__).resolve().parent.parent / "web" / "dist" | |
| BUSY_MESSAGE = "The server is busy building other decks. Try again in a moment." | |
| class Admission: | |
| """Counts completions running or waiting, and turns away the excess.""" | |
| def __init__(self, max_waiting: int): | |
| self.limit = max_waiting + 1 # the running one plus the waiting ones | |
| self.active = 0 | |
| self._lock = threading.Lock() | |
| def enter(self) -> bool: | |
| with self._lock: | |
| if self.active >= self.limit: | |
| return False | |
| self.active += 1 | |
| return True | |
| def leave(self) -> None: | |
| with self._lock: | |
| self.active -= 1 | |
| class LockedCard(BaseModel): | |
| name: str = Field(min_length=1, max_length=200) | |
| quantity: int = Field(default=1, ge=1, le=99) | |
| source: Literal["input", "synergy", "staple", "basic"] = "synergy" | |
| # False: a card the reroll keeps without the user locking it (their basics, | |
| # or the rest of a deck over 100). Only locked cards are safe from cuts. | |
| locked: bool = True | |
| class CompleteRequest(BaseModel): | |
| commanders: list[str] = Field(min_length=1, max_length=2) | |
| decklist: str = Field(default="", max_length=20_000) | |
| target_lands: int | None = Field(default=None, ge=MIN_TARGET_LANDS, le=60) | |
| prune: bool = True | |
| # Over 100 is allowed: a hand-edited deck can be bigger, and gets cut to 100. | |
| locked: list[LockedCard] = Field(default_factory=list, max_length=250) | |
| exclude: list[str] = Field(default_factory=list, max_length=1000) | |
| def create_app(builder: DeckBuilder | None = None) -> FastAPI: | |
| """`builder` lets tests inject an already-loaded DeckBuilder; otherwise | |
| one is created at startup, after syncing the model files with the Hub | |
| (server/model_files.py: installs a newer release if there is one).""" | |
| async def lifespan(app: FastAPI): | |
| if builder is None: | |
| version = sync_model_files(check_remote=update_check_enabled()) | |
| prices = PriceBook(allow_download=downloads_enabled()) | |
| stop_prices = prices.start_schedule() # background: startup never waits on it | |
| builder_ = DeckBuilder(version, prices=prices) | |
| builder_.card_stats = ensure_card_stats( | |
| (card.oracle_id for card in builder_.tok.card_by_token.values()), | |
| allow_download=update_check_enabled(), | |
| ) | |
| app.state.builder = builder_ | |
| else: | |
| app.state.builder = builder | |
| stop_prices = None | |
| yield | |
| if stop_prices: | |
| stop_prices.set() | |
| docs = os.environ.get("MTG_API_DOCS", "1") != "0" | |
| app = FastAPI( | |
| title="MTG Deck Completion", lifespan=lifespan, | |
| docs_url="/docs" if docs else None, redoc_url="/redoc" if docs else None, | |
| openapi_url="/openapi.json" if docs else None, | |
| ) | |
| app.add_middleware(GZipMiddleware, minimum_size=1000) | |
| admission = app.state.admission = Admission(int(os.environ.get("MTG_MAX_WAITING", "4"))) | |
| async def headers(request: Request, call_next): | |
| response = await call_next(request) | |
| response.headers["X-Content-Type-Options"] = "nosniff" | |
| response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" | |
| # No X-Frame-Options: Hugging Face Spaces shows the app in an iframe. | |
| path = request.url.path | |
| if path.startswith("/assets/"): | |
| # Vite puts a content hash in every asset's name. | |
| response.headers["Cache-Control"] = "public, max-age=31536000, immutable" | |
| elif not path.startswith("/api/") and response.headers.get("content-type", "").startswith("text/html"): | |
| response.headers["Cache-Control"] = "no-cache" # always pick up a new build | |
| return response | |
| async def deck_input_error(_request, exc: DeckInputError): | |
| return JSONResponse(status_code=422, content={"error": exc.message, "unresolved": exc.unresolved}) | |
| def health(): | |
| builder = app.state.builder | |
| return { | |
| "status": "ok", | |
| "device": str(builder.device), | |
| "model": builder.version, | |
| "prices_updated": builder.prices.built_at, | |
| } | |
| def config(): | |
| return { | |
| "default_target_lands": DEFAULT_TARGET_LANDS, | |
| "min_target_lands": MIN_TARGET_LANDS, | |
| "prices_updated": app.state.builder.prices.built_at, | |
| } | |
| def search_cards( | |
| q: str = Query(min_length=1, max_length=100), | |
| commander: bool = False, | |
| limit: int = Query(default=10, ge=1, le=25), | |
| colors: str | None = Query(default=None, max_length=5, pattern="^[WUBRG]*$"), | |
| basics: bool = False, | |
| ): | |
| """`colors`: only cards within this color identity ("" = colorless); | |
| `basics`: also offer basic lands (for adding cards to a deck).""" | |
| return {"cards": app.state.builder.search(q, commander_only=commander, limit=limit, colors=colors, basics=basics)} | |
| def complete(req: CompleteRequest): | |
| if not admission.enter(): | |
| return JSONResponse(status_code=503, content={"error": BUSY_MESSAGE}, headers={"Retry-After": "5"}) | |
| try: | |
| completion = app.state.builder.complete( | |
| req.commanders, req.decklist, target_lands=req.target_lands, prune=req.prune, | |
| locked=[(c.name, c.quantity, c.source, c.locked) for c in req.locked], exclude=req.exclude, | |
| ) | |
| finally: | |
| admission.leave() | |
| return completion.to_dict() | |
| if WEB_DIST.is_dir(): | |
| app.mount("/assets", StaticFiles(directory=WEB_DIST / "assets"), name="assets") | |
| def spa(path: str): | |
| file = (WEB_DIST / path).resolve() | |
| if path and file.is_file() and file.is_relative_to(WEB_DIST): | |
| return FileResponse(file) | |
| return FileResponse(WEB_DIST / "index.html") | |
| return app | |
| app = create_app() | |