feat(api): POST /pipeline/{bbb,eeg,mri} dispatch routes

src/api/main.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 
 from fastapi import FastAPI
 
+from src.api.routes import router as pipeline_router
 from src.api.schemas import HealthResponse
 
 app = FastAPI(
@@ -14,6 +15,8 @@ app = FastAPI(
     version="0.4.0",
 )
 
+app.include_router(pipeline_router)
+
 
 @app.get("/health", response_model=HealthResponse)
 def health() -> HealthResponse:

src/api/routes.py

@@ -0,0 +1,110 @@
+"""POST /pipeline/{bbb,eeg,mri} routes — thin dispatchers over the pipelines.
+
+Each route validates its request body via Pydantic, invokes the pipeline,
+reads back the produced Parquet to populate row/column counts, and returns
+a uniform PipelineResponse. Pipeline-domain errors map to standard HTTP
+codes: FileNotFoundError -> 404, ValueError -> 400, anything else -> 500.
+"""
+from __future__ import annotations
+
+import time
+from pathlib import Path
+from typing import Callable
+
+import mlflow
+import pandas as pd
+from fastapi import APIRouter, HTTPException
+
+from src.api.schemas import (
+    BBBRequest,
+    EEGRequest,
+    MRIRequest,
+    PipelineResponse,
+)
+from src.core.logger import get_logger
+from src.pipelines import bbb_pipeline, eeg_pipeline, mri_pipeline
+
+logger = get_logger(__name__)
+router = APIRouter(prefix="/pipeline")
+
+
+def _wrap(
+    experiment_name: str,
+    output_path: Path,
+    fn: Callable[[], None],
+) -> PipelineResponse:
+    """Run `fn()` (the pipeline call), gather metrics, return PipelineResponse."""
+    started = time.perf_counter()
+    try:
+        fn()
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except (ValueError, KeyError) as e:
+        # KeyError: MRI pipeline raises this when sites_csv is missing a
+        # site assignment for a subject — a user data problem, not a 500.
+        raise HTTPException(status_code=400, detail=str(e))
+    duration_sec = time.perf_counter() - started
+
+    df = pd.read_parquet(output_path)
+    runs = mlflow.search_runs(
+        experiment_names=[experiment_name],
+        max_results=1,
+        order_by=["start_time DESC"],
+    )
+    run_id = runs.iloc[0]["run_id"] if len(runs) else None
+
+    return PipelineResponse(
+        status="ok",
+        output_path=str(output_path),
+        rows=len(df),
+        columns=df.shape[1],
+        duration_sec=duration_sec,
+        mlflow_run_id=run_id,
+    )
+
+
+@router.post("/bbb", response_model=PipelineResponse)
+def run_bbb(req: BBBRequest) -> PipelineResponse:
+    """Run the BBB pipeline; return rows/cols/duration + the MLflow run id."""
+    return _wrap(
+        "bbb_pipeline",
+        Path(req.output_path),
+        lambda: bbb_pipeline.run_pipeline(
+            input_path=Path(req.input_path),
+            output_path=Path(req.output_path),
+            smiles_col=req.smiles_col,
+            n_bits=req.n_bits,
+            radius=req.radius,
+        ),
+    )
+
+
+@router.post("/eeg", response_model=PipelineResponse)
+def run_eeg(req: EEGRequest) -> PipelineResponse:
+    """Run the EEG pipeline; return rows/cols/duration + the MLflow run id."""
+    return _wrap(
+        "eeg_pipeline",
+        Path(req.output_path),
+        lambda: eeg_pipeline.run_pipeline(
+            input_path=Path(req.input_path),
+            output_path=Path(req.output_path),
+            epoch_duration_s=req.epoch_duration_s,
+            eog_ch_name=req.eog_ch_name,
+            n_components=req.n_components,
+            random_state=req.random_state,
+        ),
+    )
+
+
+@router.post("/mri", response_model=PipelineResponse)
+def run_mri(req: MRIRequest) -> PipelineResponse:
+    """Run the MRI pipeline; return rows/cols/duration + the MLflow run id."""
+    return _wrap(
+        "mri_pipeline",
+        Path(req.output_path),
+        lambda: mri_pipeline.run_pipeline(
+            input_dir=Path(req.input_dir),
+            sites_csv=Path(req.sites_csv),
+            output_path=Path(req.output_path),
+        ),
+    )

tests/api/test_routes.py

@@ -0,0 +1,72 @@
+"""Tests for /pipeline/{bbb,eeg,mri} POST endpoints."""
+from __future__ import annotations
+
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from src.api.main import app
+
+
+client = TestClient(app)
+_FIXTURES = Path(__file__).resolve().parents[1] / "fixtures"
+
+
+class TestBBBRoute:
+    def test_returns_200_with_valid_input(self, tmp_path: Path):
+        out = tmp_path / "out.parquet"
+        resp = client.post(
+            "/pipeline/bbb",
+            json={
+                "input_path": str(_FIXTURES / "bbbp_sample.csv"),
+                "output_path": str(out),
+            },
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["status"] == "ok"
+        assert body["rows"] > 0
+        assert out.exists()
+
+    def test_returns_404_when_input_missing(self, tmp_path: Path):
+        resp = client.post(
+            "/pipeline/bbb",
+            json={
+                "input_path": str(tmp_path / "does_not_exist.csv"),
+                "output_path": str(tmp_path / "out.parquet"),
+            },
+        )
+        assert resp.status_code == 404
+
+    def test_returns_422_on_malformed_body(self):
+        resp = client.post("/pipeline/bbb", json={"banana": 1})
+        assert resp.status_code == 422  # pydantic validation
+
+
+class TestEEGRoute:
+    def test_returns_200_with_valid_input(self, tmp_path: Path):
+        fif = _FIXTURES / "eeg_sample.fif"
+        out = tmp_path / "out.parquet"
+        resp = client.post(
+            "/pipeline/eeg",
+            json={"input_path": str(fif), "output_path": str(out)},
+        )
+        assert resp.status_code == 200
+        assert resp.json()["rows"] > 0
+
+
+class TestMRIRoute:
+    def test_returns_200_with_valid_input(self, tmp_path: Path):
+        from tests.fixtures.build_mri_fixture import build as build_mri
+        fixture_dir = build_mri(out_dir=tmp_path / "mri_fixture")
+        out = tmp_path / "out.parquet"
+        resp = client.post(
+            "/pipeline/mri",
+            json={
+                "input_dir": str(fixture_dir),
+                "sites_csv": str(fixture_dir / "sites.csv"),
+                "output_path": str(out),
+            },
+        )
+        assert resp.status_code == 200
+        assert resp.json()["rows"] > 0