Backend Code Walkthrough¶

This page is a guided tutorial through the Mirabelle backend codebase for developers learning the system. It covers design decisions, patterns, and annotated examples.

The Single Large `main.py` — Design Decision¶

files/api/main.py is a 28000+ line file containing 200+ API endpoints. This is intentional:

Why one large file?

Single source of truth: All API behaviour is in one place. There is no hunting across sub-modules to understand what an endpoint does. New developers can use IDE search to find any endpoint immediately.
Shared state is straightforward: The in-memory _account_cache dict, the _process_registry dict (tracking running pipeline processes), and the data cache are module-level globals. In a split architecture, these would require inter-module imports or a shared state module, adding complexity.
FastAPI's dependency injection scales well: The Depends(get_current_user) pattern means auth is cleanly separated without requiring routers. Adding a new endpoint is three lines.

What is in sub-files?

Three logical areas are separated into their own files: - files/api/auth.py — authentication routes (/api/auth/*) — separated because it has its own internal state (password hashing context, token creation) and is used by the middleware - files/api/admin.py — SuperAdmin account provisioning (/api/admin/*) — separated because it operates on the master DB, not tenant DBs - files/api/supply_router.py — supply planning routes (/supply/*) — separated due to sheer size of the supply domain

FastAPI App Setup¶

app = FastAPI(
    title="Time Series Forecasting API",
    description="API for accessing forecasts, demand analytics, and MEIO data",
    version="1.1.0",
    swagger_ui_parameters={"persistAuthorization": True},
)

persistAuthorization=True keeps the JWT token populated between Swagger UI page refreshes — essential for development.

OpenAPI Security Schema¶

A custom openapi() function injects the Bearer auth scheme into the OpenAPI schema so Swagger UI shows an "Authorize" button:

def _custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    schema = get_openapi(title=app.title, version=app.version, ...)
    schema["components"]["securitySchemes"]["BearerAuth"] = {
        "type": "http",
        "scheme": "bearer",
        "bearerFormat": "JWT",
    }
    schema["security"] = [{"BearerAuth": []}]
    app.openapi_schema = schema
    return schema

app.openapi = _custom_openapi

Middleware Stack (Order Matters)¶

# 1. Timing middleware (logs request start/end + elapsed time)
app.add_middleware(_TimingMiddleware)

# 2. CORS (must be before JWT so OPTIONS preflight passes)
app.add_middleware(CORSMiddleware, allow_origins=[...], ...)

# 3. JWT auth (added last — runs first due to Starlette's LIFO middleware stack)
app.add_middleware(JWTAuthMiddleware, secret_key=..., account_cache=_account_cache)

Middleware ordering in Starlette

Starlette middleware is applied in reverse order (LIFO). app.add_middleware() prepends each middleware to the chain. So the last add_middleware() call is the outermost middleware that runs first. JWT auth is added last so it runs outermost and validates before any other logic.

Router Inclusion¶

from api.auth import router as auth_router
app.include_router(auth_router)          # prefix: /api/auth

from api.admin import router as admin_router
app.include_router(admin_router)         # prefix: /api/admin

from api.supply_router import router as supply_router
app.include_router(supply_router)        # prefix: /supply (note: no /api prefix)

Sub-routers are wrapped in try/except blocks so the API starts even if an optional dependency (e.g. a missing compiled .pyd module) fails to import. A logging.warning() is emitted but the server continues.

The `_account_cache` — Tenant DB Lookup¶

_account_cache: dict = {}

def _build_account_cache() -> None:
    global _account_cache
    for acc in get_all_accounts():       # queries master.accounts
        cp = acc.get("connection_params") or {}
        _account_cache[acc["id"]] = {
            "host":     cp.get("host", master_pg["host"]),
            "database": acc["db_name"],
            "schema":   acc["schema_name"],
            # ...
        }

_build_account_cache()  # runs at module import time (server startup)

# Share the same dict object with admin.py — mutations are immediately visible
if _ADMIN_AVAILABLE:
    set_account_cache_ref(_account_cache)

Key design: The dict is passed by reference to JWTAuthMiddleware and admin.py. When _build_account_cache() calls _account_cache.clear() + _account_cache.update(new_entries), the middleware instantly sees the updated contents without any restart or notification mechanism.

DB Connection Pattern¶

Every endpoint that needs a database connection follows the same pattern:

@app.get("/api/forecasts")
async def get_forecasts(pipeline_id: int, user: dict = Depends(get_current_user)):
    schema = get_schema()           # returns 'scenario' (or tenant-specific schema)
    conn = get_conn()               # opens new psycopg2 connection to tenant DB
    try:
        with conn.cursor() as cur:
            cur.execute(
                f"SELECT item_id, site_id, method, point_forecast "
                f"FROM {schema}.forecast_results "
                f"WHERE pipeline_id = %s",
                (pipeline_id,)
            )
            rows = cur.fetchall()
            cols = [d[0] for d in cur.description]
            results = [dict(zip(cols, row)) for row in rows]
        conn.commit()
        return results
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        conn.close()                # always close — no pool

Points to note: 1. get_schema() reads from _account_ctx (set by middleware) or falls back to config.yaml. 2. get_conn() reads from _account_ctx for the actual DB credentials. 3. conn.autocommit = False (set in get_conn()) — all writes are in explicit transactions. 4. conn.close() in finally — no connection pool, every request opens and closes a connection. 5. cur.description is used to build column-name dicts from raw tuples.

Schema Injection — Security Note¶

All table references use string formatting: f"FROM {schema}.forecast_results". The schema value comes from get_schema() which reads from the auth context or config.yaml. Since it is never derived from user input, this is safe. Do not use f-string formatting for any user-supplied values — always use %s parameterized queries.

Dependency Injection for Auth¶

The get_current_user function is defined in files/api/auth.py and imported into main.py:

from api.auth import get_current_user

def get_current_user(request: Request) -> dict:
    """Read the user dict that JWTAuthMiddleware attached to request.state."""
    user = getattr(request.state, "user", None)
    if user is None:
        raise HTTPException(
            status_code=401,
            detail="Not authenticated",
        )
    return user

Used in every protected endpoint:

@app.get("/api/series/{unique_id}")
async def get_series(
    unique_id: str,
    pipeline_id: int,
    user: dict = Depends(get_current_user),  # inject authenticated user
):
    ...

FastAPI calls get_current_user(request) before the handler function body executes. If the user is not authenticated, FastAPI returns 401 before the handler runs.

The `load_config_from_db()` Pattern¶

Rather than reading configuration from config.yaml, pipeline steps call load_config_from_db() from files/db/db.py:

def load_config_from_db() -> dict:
    """
    Load all default parameter sets from the DB and return them as a
    unified config dict keyed by parameter_type.

    Example return:
    {
        "outlier_detection": {"method": "iqr", "threshold": 1.5, ...},
        "forecasting": {"horizon": 78, "models": ["AutoARIMA", "AutoETS"], ...},
        "meio": {"distribution_families": ["normal", "gamma", ...], ...},
        ...
    }
    """
    conn = get_conn()
    try:
        schema = get_schema()
        with conn.cursor() as cur:
            cur.execute(
                f"SELECT parameter_type, parameters_set FROM {schema}.parameters "
                f"WHERE is_default = TRUE"
            )
            return {row[0]: row[1] for row in cur.fetchall()}
    finally:
        conn.close()

This replaces the anti-pattern of having configuration in files. All pipeline parameters — thresholds, model lists, scoring weights, distribution families — live in the scenario.parameters table and can be updated through the Settings UI without deploying code.

The ParameterResolver (3-Level Hierarchy)¶

files/utils/parameter_resolver.py implements parameter resolution for pipeline steps:

Level 1 (most specific): series_parameter_assignment
  → The SKU has its own per-step parameter override

Level 2: segment membership + parameter_segment
  → The SKU belongs to a segment that has a parameter set assigned

Level 3 (default): parameters WHERE is_default = TRUE
  → The global default for this parameter_type

This allows a "high-volatility intermittent" segment to use different outlier thresholds than a "steady state" segment, while individual critical SKUs can have their own fully custom configuration.

A Representative Endpoint — Annotated¶

Here is a representative endpoint from main.py showing all patterns:

@app.get("/api/forecast/{unique_id}/best")
async def get_best_method_for_series(
    unique_id: str,                              # path parameter
    pipeline_id: int = Query(...),               # required query parameter
    scenario_id: int = Query(1),                 # optional query parameter with default
    user: dict = Depends(get_current_user),      # auth injection
):
    """Return the best forecasting method for a specific series."""

    schema = get_schema()
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            # Parameterized query — no f-string for user input
            cur.execute(
                f"""
                SELECT
                    bm.item_id, bm.site_id,
                    bm.best_method,
                    bm.best_score,
                    bm.runner_up_method,
                    bm.locked_method,
                    bm.locked_by,
                    bm.locked_at
                FROM {schema}.series_best_methods bm
                WHERE (bm.item_id, bm.site_id) = %s
                  AND bm.pipeline_id = %s
                  AND bm.scenario_id = %s
                LIMIT 1
                """,
                (unique_id, pipeline_id, scenario_id)
            )
            row = cur.fetchone()
            if not row:
                raise HTTPException(status_code=404, detail="Series not found")
            cols = [d[0] for d in cur.description]
            result = dict(zip(cols, row))

        # Effective method: locked_method overrides best_method
        result["effective_method"] = result.get("locked_method") or result.get("best_method")
        return result

    except HTTPException:
        raise                           # re-raise 404/403 without wrapping
    except Exception as e:
        logger.error("Error in get_best_method: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        conn.close()                    # always release connection

Process Logging System¶

Long-running pipeline steps use ProcessLogger from files/utils/process_logger.py:

class ProcessLogger:
    def __init__(self, run_id: str, step_name: str, pipeline_id: int, scenario_id: int):
        self.run_id = run_id
        self.step_name = step_name
        self.pipeline_id = pipeline_id
        self.scenario_id = scenario_id
        self._start_time = None

    def start(self):
        self._start_time = datetime.utcnow()
        # INSERT INTO scenario.process_log (run_id, step_name, status='running', started_at, ...)
        self._write_db(status="running")

    def finish(self, rows_processed: int = 0):
        duration = (datetime.utcnow() - self._start_time).total_seconds()
        self._write_db(status="completed", duration_s=duration, rows_processed=rows_processed)

    def fail(self, error: str):
        self._write_db(status="failed", error_message=error)

Live Log Streaming (`ListHandler`)¶

The ListHandler class is a Python logging handler that buffers log lines in memory. The frontend subscribes to a Server-Sent Events endpoint that streams from this buffer:

class ListHandler(logging.Handler):
    def __init__(self):
        super().__init__()
        self.records: list[str] = []

    def emit(self, record: logging.LogRecord):
        self.records.append(self.format(record))

When a pipeline step is triggered via the API:

A run_id UUID is generated.
A ListHandler is attached to the step's logger.
The step runs in a background thread (via asyncio.create_task or threading.Thread).
The frontend polls GET /api/process/stream/{run_id}?token=<jwt> using EventSource.
The SSE handler reads from handler.records and sends each line as an SSE event.

@app.get("/api/process/stream/{run_id}")
async def stream_process_log(run_id: str, token: str = Query(...)):
    # token validated by middleware via query param
    handler = _process_registry.get(run_id)  # dict of run_id → ListHandler

    async def event_generator():
        sent = 0
        while True:
            records = handler.records[sent:]
            for line in records:
                yield f"data: {json.dumps({'line': line})}\n\n"
            sent += len(records)
            if handler.done:
                break
            await asyncio.sleep(0.5)

    return StreamingResponse(event_generator(), media_type="text/event-stream")

Dask Orchestration¶

Forecasting runs use Dask for parallelisation. The orchestrator in files/utils/orchestrator.py partitions the series list and maps forecasting functions across Dask workers.

import dask.dataframe as dd
from dask.distributed import Client

def run_parallel_forecasting(series_list: list, config: dict) -> list:
    scheduler = config.get("parallel", {}).get("scheduler", "synchronous")

    if scheduler == "synchronous":
        # Single-threaded fallback for debugging
        return [_forecast_one_series(s, config) for s in series_list]

    with Client(n_workers=config.get("parallel", {}).get("n_workers", 4)) as client:
        futures = client.map(_forecast_one_series, series_list, config=config)
        return client.gather(futures)

Dask and the account context

Dask workers run in separate OS processes. They do not inherit the FastAPI request context (no _account_ctx ContextVar). Pipeline scripts must pass DB credentials explicitly via environment variables when running in distributed Dask mode.

Error Handling Conventions¶

All endpoints follow this pattern:

Re-raise HTTPException directly (never wrap a 404 in a 500).
Log unexpected exceptions with logger.error(..., exc_info=True) for stack traces.
Rollback the DB transaction on any exception.
Always close the DB connection in finally.

try:
    # ... database work ...
    conn.commit()
    return result
except HTTPException:
    raise                           # 404, 403, 422 pass through unmodified
except psycopg2.IntegrityError as e:
    conn.rollback()
    raise HTTPException(status_code=409, detail=f"Conflict: {e}")
except Exception as e:
    conn.rollback()
    logger.error("Unexpected error: %s", e, exc_info=True)
    raise HTTPException(status_code=500, detail=str(e))
finally:
    conn.close()

Supply API Rule¶

Every call to a /supply/* endpoint must include both pipeline_id and scenario_id as query parameters.

Why: The supply planning engine is fully scenario-scoped. Every supply run (supply_run), every order (supply_order), every projection (supply_inventory_projection) is tagged with both IDs. Without them, the backend cannot resolve which data to return.

On the frontend, always use the useSupplyParams() hook from files/frontend/src/utils/useSupplyParams.js:

import { useSupplyParams } from '../utils/useSupplyParams';

function MyComponent() {
    const supplyParams = useSupplyParams();
    // supplyParams = { pipeline_id: 1, scenario_id: 2 }

    useEffect(() => {
        api.get('/supply/inventory', {
            params: { ...supplyParams, run_id: rid, sku_id: x }
        });
    }, [supplyParams]);
}

Exceptions — the following endpoints are immutable master-data endpoints and do not need pipeline/scenario: - /supply/item-site — item-site cost parameters - /supply/sales-returns CRUD — return logistics

OpenTelemetry Integration¶

main.py optionally instruments the FastAPI app with OpenTelemetry for distributed tracing and structured logging:

try:
    from api.otel_setup import init_otel, get_tracer, span as otel_span
    _OTEL_SETUP_AVAILABLE = True
except Exception:
    _OTEL_SETUP_AVAILABLE = False
    # Stubs so the rest of the file can use otel_span() unconditionally
    from contextlib import contextmanager
    @contextmanager
    def otel_span(name, attributes=None):
        yield None

When OTel is available, init_otel(app) instruments all FastAPI routes automatically. Within handlers, with otel_span("forecast.query", {"unique_id": uid}) creates child spans. The stubs ensure the same code runs in both OTel and non-OTel environments.

Parameter Registry¶

main.py defines PARAMETER_REGISTRY — a metadata list that drives the Settings UI:

PARAMETER_REGISTRY = [
    {"parameter_type": "data_source",       "label": "ETL",               "description": "..."},
    {"parameter_type": "outlier_detection", "label": "Outlier Detection",  "description": "..."},
    {"parameter_type": "characterization",  "label": "Characterization",   "description": "..."},
    {"parameter_type": "forecasting",       "label": "Forecasting",        "description": "..."},
    {"parameter_type": "backtesting",       "label": "Backtesting",        "description": "..."},
    {"parameter_type": "meio",              "label": "MEIO",               "description": "..."},
    {"parameter_type": "parallel",          "label": "Parallel",           "description": "..."},
    {"parameter_type": "auth",              "label": "Auth",               "description": "..."},
    {"parameter_type": "segmentation",      "label": "Segmentation",       "description": "..."},
    {"parameter_type": "supply",            "label": "Supply",             "description": "..."},
]

GET /api/parameters/registry returns this list. The Settings UI uses it to generate one tab per parameter type and fetches the actual JSONB parameter sets from GET /api/parameters/{type}.

Adding a New Endpoint — Step by Step¶

Find the right location in main.py — endpoints are grouped by domain (search for the section comment, e.g. # ── MEIO endpoints ──).
Write the handler:
```
                  
```
href="#__codelineno-19-1">@app.get("/api/my-domain/my-endpoint") class="k">async def my_endpoint( pipeline_id: int = Query(...), user: dict = Depends(get_current_user), class="p">): schema = get_schema() conn = get_conn() try: with conn.cursor() as cur: cur.execute( f"SELECT ... FROM {schema}.my_table WHERE pipeline_id = %s", (pipeline_id,) ) rows = cur.fetchall() cols = [d[0] for d in cur.description] return [dict(zip(cols, row)) for row in rows] except Exception as e: conn.rollback() raise HTTPException(status_code=500, detail=str(e)) finally: conn.close()

Add a Pydantic model if the endpoint accepts a request body:

class MyRequest(BaseModel):
    field_one: str
    field_two: Optional[int] = None

@app.post("/api/my-domain/my-endpoint")
async def create_something(
    body: MyRequest,
    user: dict = Depends(get_current_user),
):
    ...

Test via Swagger UI at http://localhost:8002/docs — click Authorize and paste your JWT token.
Add a corresponding frontend call in the appropriate component using the shared api Axios instance from files/frontend/src/utils/api.js.