SupplyAgent/api/routers/pipeline.py

from __future__ import annotations

from fastapi import APIRouter, HTTPException, Query

from api.schemas.pipeline import CreatePipelineRunBody
from api.services.pipeline import (
    cancel_run,
    create_run,
    get_run,
    list_runs,
    pipeline_health,
    resume_run,
    retry_step,
)

router = APIRouter(prefix="/api/pipeline", tags=["pipeline"])


@router.get("/runs")
def pipeline_runs(
    limit: int = Query(default=50, ge=1, le=200),
    status: str | None = None,
    biz_dt: str | None = Query(default=None, pattern=r"^\d{8}$"),
) -> dict:
    return {"items": list_runs(limit=limit, status=status, biz_dt=biz_dt)}


@router.post("/runs", status_code=202)
def create_pipeline_run(body: CreatePipelineRunBody | None = None) -> dict:
    payload = body or CreatePipelineRunBody()
    return create_run(
        biz_dt=payload.biz_dt,
        source="api",
        reason=payload.reason,
    )


@router.get("/runs/{run_id}")
def pipeline_run_detail(run_id: str) -> dict:
    result = get_run(run_id)
    if result is None:
        raise HTTPException(status_code=404, detail="pipeline run not found")
    return result


@router.get("/runs/{run_id}/steps")
def pipeline_run_steps(run_id: str) -> dict:
    result = get_run(run_id)
    if result is None:
        raise HTTPException(status_code=404, detail="pipeline run not found")
    return {"items": result["steps"]}


@router.post("/runs/{run_id}/resume", status_code=202)
def resume_pipeline_run(run_id: str) -> dict:
    if not resume_run(run_id):
        raise HTTPException(status_code=404, detail="pipeline run not found or not resumable")
    result = get_run(run_id)
    assert result is not None
    return {"accepted": True, "run_id": run_id, "status": result["status"]}


@router.post("/runs/{run_id}/cancel", status_code=202)
def cancel_pipeline_run(run_id: str) -> dict:
    if not cancel_run(run_id):
        raise HTTPException(status_code=404, detail="pipeline run not found")
    result = get_run(run_id)
    assert result is not None
    return {"accepted": True, "run_id": run_id, "status": result["status"]}


@router.post("/runs/{run_id}/steps/{step_key}/retry", status_code=202)
def retry_pipeline_run_step(run_id: str, step_key: str) -> dict:
    if not retry_step(run_id, step_key):
        raise HTTPException(
            status_code=409,
            detail="step is not the first failed step or run is not resumable",
        )
    return {
        "accepted": True,
        "run_id": run_id,
        "step_key": step_key,
        "status": "queued",
    }


@router.get("/health")
def health() -> dict:
    return pipeline_health()