Richie
add7a6a848
treefmt / nix fmt (pull_request) Successful in 6s
test ebook search / test-ebook-search (pull_request) Successful in 39s
pytest / pytest (pull_request) Successful in 40s
build_systems / build-brain (pull_request) Successful in 1m0s
build_systems / build-bob (pull_request) Successful in 1m2s
build_systems / build-rhapsody-in-green (pull_request) Successful in 1m33s
build_systems / build-jeeves (pull_request) Successful in 2m25s
test ebook search / test-ebook-search (push) Successful in 35s
build_systems / build-brain (push) Successful in 37s
build_systems / build-bob (push) Successful in 37s
build_systems / build-jeeves (push) Successful in 2m17s
treefmt / nix fmt (push) Successful in 5s
pytest / pytest (push) Successful in 29s
build_systems / build-rhapsody-in-green (push) Successful in 49s
- implement FastAPI and HTMX lobby and game interfaces - support up to four human and AI-controlled players - add configurable rules, victory conditions, and expansion modules - support validated JSON cards, patrons, objectives, and outposts
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
"""FastAPI entry point for Gems."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from contextlib import asynccontextmanager, suppress
|
|
from typing import TYPE_CHECKING, Annotated
|
|
|
|
import typer
|
|
import uvicorn
|
|
from fastapi import FastAPI, Request, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from .config import load_config
|
|
from .events import EventBroker
|
|
from .persistence import Repository
|
|
from .rooms import RoomService
|
|
from .routes import router
|
|
from .security import CredentialService
|
|
from .web import STATIC_DIR
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator
|
|
|
|
from starlette.middleware.base import RequestResponseEndpoint
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|
"""Initialize and close application-scoped services."""
|
|
config = app.state.config
|
|
repository = Repository(config.database_path)
|
|
app.state.repository = repository
|
|
app.state.rooms = RoomService(repository, CredentialService(config.key_path), EventBroker())
|
|
repository.cleanup()
|
|
cleanup_task = asyncio.create_task(_cleanup_rooms(repository))
|
|
try:
|
|
yield
|
|
finally:
|
|
cleanup_task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await cleanup_task
|
|
repository.close()
|
|
|
|
|
|
async def _cleanup_rooms(repository: Repository) -> None:
|
|
while True:
|
|
await asyncio.sleep(3600)
|
|
repository.cleanup()
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
"""Create an isolated application instance."""
|
|
app = FastAPI(title="Gems", docs_url=None, redoc_url=None, lifespan=lifespan)
|
|
app.state.config = load_config()
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
app.include_router(router)
|
|
|
|
@app.middleware("http")
|
|
async def security_headers(request: Request, call_next: RequestResponseEndpoint) -> Response:
|
|
response = await call_next(request)
|
|
response.headers.setdefault("Referrer-Policy", "no-referrer")
|
|
response.headers.setdefault("X-Content-Type-Options", "nosniff")
|
|
response.headers.setdefault("X-Frame-Options", "DENY")
|
|
response.headers.setdefault("X-Robots-Tag", "noindex, nofollow")
|
|
response.headers.setdefault(
|
|
"Content-Security-Policy",
|
|
"default-src 'self'; script-src 'self'; style-src 'self'; style-src-attr 'unsafe-inline'; "
|
|
"img-src 'self' data:; connect-src 'self'",
|
|
)
|
|
if request.url.path.startswith("/rooms/"):
|
|
response.headers.setdefault("Cache-Control", "no-store")
|
|
return response
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
def serve(
|
|
host: Annotated[str | None, typer.Option()] = None,
|
|
port: Annotated[int | None, typer.Option()] = None,
|
|
) -> None:
|
|
"""Run the Gems ASGI application with Uvicorn."""
|
|
config = load_config()
|
|
uvicorn.run("python.gems.main:app", host=host or config.host, port=port or config.port, workers=1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
typer.run(serve)
|