"""
Scheduler Service — FastAPI + Graphene GraphQL application.
A Python GraphQL API that replicates the PHP GraphQL scheduler widget endpoints.
Supports dynamic per-request database connections via the X-DB-Name header,
and JWT / X-Token authentication.
GraphQL endpoint: POST /
GraphiQL IDE: GET / (browser)
Health check: GET /health
"""
import json
import logging
from typing import Optional
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse
from schema import schema
from auth import authenticate
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("scheduler")
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
app = FastAPI(
title="MYL Scheduler GraphQL API",
description="Python GraphQL API for league game scheduling",
version="1.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=[], # Restricted — Laravel proxy handles CORS
allow_methods=["GET", "POST"],
allow_headers=["*"],
)
# ---------------------------------------------------------------------------
# Context getter — runs before every GraphQL request
# ---------------------------------------------------------------------------
async def get_context(request: Request) -> dict:
"""
Extract and validate the X-DB-Name header and authenticate the user.
The returned dict is available as `info.context` in all resolvers.
Bypasses header and auth requirements for introspection and playground GET requests.
"""
is_introspection = False
if request.method == "GET":
is_introspection = True
elif request.method == "POST":
content_type = request.headers.get("content-type", "")
if "application/json" in content_type:
try:
body = await request.json()
query = body.get("query", "")
if "__schema" in query or "IntrospectionQuery" in query:
is_introspection = True
except Exception:
pass
if is_introspection:
return {
"request": request,
"db_name": "",
"user": None,
}
# 1. Require database name header
db_name = (
request.headers.get("X-Organisation")
or request.headers.get("X-DB-Name")
or request.headers.get("db-name")
or request.headers.get("X-DB")
)
if not db_name:
raise HTTPException(
status_code=400,
detail="Organisation header (X-Organisation) is required.",
)
# 2. Authenticate
user = authenticate(request)
return {
"request": request,
"db_name": db_name,
"user": user,
}
# ---------------------------------------------------------------------------
# Health check (unauthenticated — must be before GraphQL router)
# ---------------------------------------------------------------------------
@app.get("/health")
def health():
"""Simple health-check endpoint for the Laravel proxy to ping."""
return {"status": "ok"}
# ---------------------------------------------------------------------------
# Graphene GraphiQL Playground & Execution Handlers
# ---------------------------------------------------------------------------
GRAPHIQL_HTML = """
GraphiQL
"""
@app.get("/")
async def graphql_playground():
"""Serve the GraphiQL playground."""
return HTMLResponse(content=GRAPHIQL_HTML)
def set_value_at_path(variables, path, value):
"""Helper to inject mapped files at specified nested variables paths."""
parts = path.split(".")
if parts[0] != "variables":
return
curr = variables
for part in parts[1:-1]:
if part not in curr or curr[part] is None:
curr[part] = {}
curr = curr[part]
curr[parts[-1]] = value
@app.post("/")
async def graphql_endpoint(request: Request):
"""Handle GraphQL queries and mutations via Graphene."""
content_type = request.headers.get("content-type", "")
# Retrieve context (runs auth / header verification unless it is an introspection query)
context = await get_context(request)
if "multipart/form-data" in content_type:
form = await request.form()
operations_str = form.get("operations")
map_str = form.get("map")
if not operations_str or not map_str:
raise HTTPException(status_code=400, detail="Invalid multipart GraphQL request")
operations = json.loads(operations_str)
map_data = json.loads(map_str)
variables = operations.get("variables") or {}
# Inject uploaded files to variables using paths mapping
for file_key, paths in map_data.items():
file_item = form.get(file_key)
if file_item:
for path in paths:
set_value_at_path(variables, path, file_item)
query = operations.get("query")
operation_name = operations.get("operationName")
else:
# Standard JSON request
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON body")
query = body.get("query")
variables = body.get("variables")
operation_name = body.get("operationName")
# Execute Graphene schema asynchronously
result = await schema.execute_async(
query,
variable_values=variables,
operation_name=operation_name,
context_value=context
)
response_data = {}
if result.data is not None:
response_data["data"] = result.data
if result.errors:
response_data["errors"] = [{"message": str(err.original_error or err)} for err in result.errors]
return JSONResponse(content=response_data)