pipekit/pipekit/api/app.py

"""FastAPI app factory.

JSON endpoints live under ``/api``. HTML pages (added in a later
increment) will live at ``/``. Keeping them separate avoids
content-negotiation complexity and keeps the API curl-testable.
"""

from __future__ import annotations

import logging
import sys
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.exception_handlers import http_exception_handler
from starlette.exceptions import HTTPException as StarletteHTTPException

import os

from .. import __version__, db, jrunner, repo
from ..web import mount_web
from .routes import connections, groups, introspect, modules, runs, system

_log = logging.getLogger("pipekit.web")


def _configure_logging() -> None:
    """Route pipekit's loggers to stderr (→ journal) with a consistent format.

    Uvicorn configures only its own loggers, not the root, so without this our
    records would fall through to the last-resort handler (WARNING+ only) and
    INFO breadcrumbs would vanish. Idempotent — safe to call per create_app."""
    base = logging.getLogger("pipekit")
    if not base.handlers:
        handler = logging.StreamHandler(sys.stderr)
        handler.setFormatter(logging.Formatter(
            "%(asctime)s %(levelname)s [%(name)s] %(message)s"))
        base.addHandler(handler)
        base.setLevel(logging.INFO)
        base.propagate = False


_SECRET_KEYS = ("password", "passwd", "pwd", "secret", "token")


def _redact_body(raw: bytes, content_type: str, limit: int = 2000) -> str:
    """Decode a captured request body for logging, masking secret-looking
    fields (passwords etc.) so they never reach the journal."""
    if not raw:
        return ""
    text = raw.decode("utf-8", "replace")
    ct = (content_type or "").lower()
    try:
        if "application/json" in ct:
            import json as _json
            obj = _json.loads(text)
            if isinstance(obj, dict):
                obj = {k: ("***" if any(s in k.lower() for s in _SECRET_KEYS) else v)
                       for k, v in obj.items()}
                text = _json.dumps(obj)
        else:
            from urllib.parse import parse_qsl, urlencode
            pairs = parse_qsl(text, keep_blank_values=True)
            if pairs:
                text = urlencode([(k, "***" if any(s in k.lower() for s in _SECRET_KEYS) else v)
                                  for k, v in pairs])
    except Exception:                                                # noqa: BLE001
        pass
    return text[:limit] + ("…[truncated]" if len(text) > limit else "")


class _RequestBodyCapture:
    """Buffer each HTTP request body onto the scope (``pk_raw_body``) so the
    exception handler can log the submitted payload on failure, then replay it
    downstream so route handlers still read the body normally. Bodies here are
    small form/JSON posts — no large uploads go through HTTP."""

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        body = bytearray()
        while True:
            message = await receive()
            if message["type"] == "http.request":
                body += message.get("body", b"")
                if not message.get("more_body", False):
                    break
            else:  # http.disconnect
                break
        scope["pk_raw_body"] = bytes(body)
        replayed = False

        async def replay():
            nonlocal replayed
            if not replayed:
                replayed = True
                return {"type": "http.request", "body": bytes(body), "more_body": False}
            return {"type": "http.disconnect"}

        await self.app(scope, replay, send)


def _live_pids() -> set[int]:
    """PIDs currently alive on this host (Linux /proc). Empty set elsewhere."""
    try:
        return {int(p) for p in os.listdir("/proc") if p.isdigit()}
    except OSError:
        return set()


@asynccontextmanager
async def _lifespan(app: FastAPI):
    # Release module locks left behind by a previous process that died mid-run
    # (a hard kill skips the engine's finally, so `running` stays 1 and future
    # runs would refuse with "already running"). Clears locks held by a dead PID
    # or older than 24h.
    try:
        n = repo.clear_stale_locks(live_pids=_live_pids())
        if n:
            _log.warning("cleared %d stale module lock(s) at startup", n)
    except Exception:                                            # noqa: BLE001
        _log.exception("stale-lock cleanup failed")
    # Finalise run_log / group_run rows abandoned by a process that died
    # mid-run — in a fresh process nothing legitimate is running yet, so any
    # `running` row is a leftover that would otherwise poll forever.
    try:
        runs, group_runs = repo.reconcile_abandoned_runs(
            "run abandoned — pipekit process exited before completion "
            "(reconciled at startup)")
        if runs or group_runs:
            _log.warning("reconciled %d abandoned run(s) and %d group run(s) "
                         "at startup", runs, group_runs)
    except Exception:                                            # noqa: BLE001
        _log.exception("abandoned-run reconciliation failed")
    from ..scheduler import start_scheduler
    start_scheduler()
    yield


def create_app() -> FastAPI:
    _configure_logging()
    app = FastAPI(title="Pipekit", version=__version__, lifespan=_lifespan)
    app.add_middleware(_RequestBodyCapture)

    @app.exception_handler(StarletteHTTPException)
    async def _logged_http_exception_handler(request: Request,
                                             exc: StarletteHTTPException):
        # FastAPI turns HTTPException into a normal response and never logs it,
        # so failures surfaced this way (e.g. wizard dest-provisioning errors)
        # were invisible in the journal. Log them — with the submitted payload,
        # secrets masked — then defer to the default handler for the response.
        body = _redact_body(request.scope.get("pk_raw_body", b""),
                            request.headers.get("content-type", ""))
        if exc.status_code >= 500:
            _log.error("%s %s -> %d: %s | body: %s", request.method,
                       request.url.path, exc.status_code, exc.detail, body,
                       exc_info=exc)
        elif exc.status_code >= 400:
            _log.warning("%s %s -> %d: %s | body: %s", request.method,
                         request.url.path, exc.status_code, exc.detail, body)
        return await http_exception_handler(request, exc)

    app.include_router(system.router)
    app.include_router(connections.router, prefix="/api")
    app.include_router(introspect.router, prefix="/api")
    app.include_router(groups.router, prefix="/api")
    app.include_router(modules.router, prefix="/api")
    app.include_router(runs.router, prefix="/api")
    mount_web(app)
    return app