Add operations overview endpoint

.env.example

@@ -1,4 +1,7 @@
 APP_NAME="DataOps Observability API"
 ENVIRONMENT=local
 DATABASE_URL=sqlite:///./dataops_observability.db
+API_PREFIX=/api/v1APP_NAME="DataOps Observability API"
+ENVIRONMENT=local
+DATABASE_URL=sqlite:///./dataops_observability.db
 API_PREFIX=/api/v1

.github/workflows/ci.yml

@@ -2,7 +2,8 @@ name: CI
 
 on:
   push:
-    branches: ["main"]
+    branches:
+      - main
   pull_request:
 
 permissions:
@@ -12,22 +13,28 @@ jobs:
   quality:
     runs-on: ubuntu-latest
     steps:
-      - name: Checkout
+      - name: Check out repository
         uses: actions/checkout@v4
-      - name: Setup Python
+
+      - name: Set up Python
         uses: actions/setup-python@v5
         with:
           python-version: "3.11"
           cache: pip
-      - name: Install
+
+      - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
           python -m pip install -e ".[dev]"
-      - name: Ruff
+
+      - name: Lint
         run: ruff check .
-      - name: Mypy
+
+      - name: Type check
         run: mypy app
-      - name: Migrations
+
+      - name: Run migrations
         run: alembic upgrade head
-      - name: Tests
-        run: pytest --cov=app --cov-report=term-missing
+
+      - name: Test
+        run: pytest

.gitignore

@@ -1,3 +1,4 @@
+.env
 .venv/
 __pycache__/
 *.py[cod]
@@ -9,6 +10,19 @@ htmlcov/
 dist/
 build/
 *.egg-info/
+*.db
+*.sqlite
+*.sqlite3.venv/
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.ruff_cache/
+.mypy_cache/
+.coverage
+htmlcov/
+dist/
+build/
+*.egg-info/
 .env
 *.db
 *.sqlite

README.md

@@ -75,6 +75,7 @@ Seed and inspect local sample data:
 python scripts/seed_sample_data.py
 Invoke-RestMethod "http://127.0.0.1:8000/api/v1/pipeline-runs/latest?name=orders_daily_load"
 Invoke-RestMethod "http://127.0.0.1:8000/api/v1/metrics/pipelines"
+Invoke-RestMethod "http://127.0.0.1:8000/api/v1/metrics/operations-overview?stale_after_minutes=60"
 Invoke-RestMethod "http://127.0.0.1:8000/api/v1/metrics/quality-checks"
 Invoke-RestMethod "http://127.0.0.1:8000/api/v1/metrics/stale-pipeline-runs?max_age_minutes=60"
 ```
@@ -101,10 +102,22 @@ The container starts the FastAPI app on port `8000`. Run migrations before produ
 | POST | `/api/v1/pipeline-runs/{run_id}/quality-checks` | Add a quality check to a run |
 | GET | `/api/v1/pipeline-runs/{run_id}/quality-checks` | List quality checks for a run |
 | GET | `/api/v1/metrics/summary` | Operational summary counts |
+| GET | `/api/v1/metrics/operations-overview?stale_after_minutes=60` | Combined operator dashboard snapshot with recommended actions |
 | GET | `/api/v1/metrics/pipelines` | Pipeline health rollups grouped by name |
 | GET | `/api/v1/metrics/quality-checks` | Quality-check counts grouped by severity and status |
 | GET | `/api/v1/metrics/stale-pipeline-runs?max_age_minutes=60` | Active pipeline runs older than the configured age threshold |
 
+## Operations Overview
+
+The `/api/v1/metrics/operations-overview` endpoint combines the service's most useful operational signals into one response for dashboards or runbooks:
+
+- `service_status`: `healthy`, `degraded`, or `attention_required`
+- `summary`: total runs, run statuses, and quality-check counts
+- `pipeline_health`: latest run state and quality-check counts by pipeline name
+- `quality_checks`: severity and status rollups
+- `stale_pipeline_runs`: active runs older than the requested threshold
+- `recommended_actions`: prioritized next steps for failed checks, failed latest runs, stale active runs, and warnings
+
 ## Configuration
 
 Settings are loaded from environment variables.

app/api/pipeline_runs.py

@@ -7,6 +7,7 @@
 from app.models.pipeline import PipelineRun
 from app.schemas.pipeline import (
     MetricsSummary,
+    OperationsOverview,
     PipelineHealthRollup,
     PipelineRunCreate,
     PipelineRunRead,
@@ -23,6 +24,7 @@
     create_quality_check,
     get_latest_pipeline_run,
     get_metrics_summary,
+    get_operations_overview,
     get_pipeline_health_rollups,
     get_pipeline_run,
     get_pipeline_run_timeline,
@@ -120,6 +122,14 @@ def summary(db: Annotated[Session, Depends(get_db)]) -> MetricsSummary:
     return get_metrics_summary(db)
 
 
+@router.get("/metrics/operations-overview", response_model=OperationsOverview)
+def operations_overview(
+    db: Annotated[Session, Depends(get_db)],
+    stale_after_minutes: Annotated[int, Query(ge=1, le=1440)] = 60,
+) -> OperationsOverview:
+    return get_operations_overview(db, stale_after_minutes=stale_after_minutes)
+
+
 @router.get("/metrics/pipelines", response_model=list[PipelineHealthRollup])
 def pipeline_health(db: Annotated[Session, Depends(get_db)]) -> list[PipelineHealthRollup]:
     return get_pipeline_health_rollups(db)

app/schemas/pipeline.py

@@ -134,6 +134,22 @@ class QualityCheckSeverityRollup(BaseModel):
     failed_checks: int
 
 
+class RecommendedAction(BaseModel):
+    priority: str
+    title: str
+    detail: str
+
+
+class OperationsOverview(BaseModel):
+    generated_at: datetime
+    service_status: str
+    summary: MetricsSummary
+    pipeline_health: list[PipelineHealthRollup]
+    quality_checks: list[QualityCheckSeverityRollup]
+    stale_pipeline_runs: list[StalePipelineRunMetric]
+    recommended_actions: list[RecommendedAction]
+
+
 class SeedSampleDataSummary(BaseModel):
     pipeline_runs_created: int
     quality_checks_created: int

app/services/pipeline_runs.py

@@ -6,6 +6,7 @@
 from app.models.pipeline import PipelineRun, QualityCheck
 from app.schemas.pipeline import (
     MetricsSummary,
+    OperationsOverview,
     PipelineHealthRollup,
     PipelineRunCreate,
     PipelineRunStatus,
@@ -15,6 +16,7 @@
     QualityCheckSeverity,
     QualityCheckSeverityRollup,
     QualityCheckStatus,
+    RecommendedAction,
     StalePipelineRunMetric,
 )
 
@@ -331,3 +333,77 @@ def get_stale_pipeline_run_metrics(
         )
 
     return sorted(stale_runs, key=lambda run: run.age_minutes, reverse=True)
+
+
+def get_operations_overview(
+    db: Session,
+    stale_after_minutes: int = 60,
+    now: datetime | None = None,
+) -> OperationsOverview:
+    generated_at = _as_utc(now or datetime.now(UTC))
+    summary = get_metrics_summary(db)
+    pipeline_health = get_pipeline_health_rollups(db)
+    quality_checks = get_quality_check_severity_rollups(db)
+    stale_pipeline_runs = get_stale_pipeline_run_metrics(
+        db,
+        max_age_minutes=stale_after_minutes,
+        now=generated_at,
+    )
+    latest_failed_pipelines = [
+        rollup.name
+        for rollup in pipeline_health
+        if rollup.latest_status == PipelineRunStatus.failed
+    ]
+
+    recommended_actions: list[RecommendedAction] = []
+    if summary.failed_quality_checks:
+        recommended_actions.append(
+            RecommendedAction(
+                priority="critical",
+                title="Investigate failed quality checks",
+                detail=f"{summary.failed_quality_checks} failed checks require review.",
+            )
+        )
+    if latest_failed_pipelines:
+        recommended_actions.append(
+            RecommendedAction(
+                priority="high",
+                title="Review failed pipeline runs",
+                detail="Latest failed pipelines: " + ", ".join(latest_failed_pipelines),
+            )
+        )
+    if stale_pipeline_runs:
+        recommended_actions.append(
+            RecommendedAction(
+                priority="high",
+                title="Resolve stale active runs",
+                detail=(
+                    f"{len(stale_pipeline_runs)} active runs exceeded "
+                    f"{stale_after_minutes} minutes."
+                ),
+            )
+        )
+    if summary.warning_quality_checks:
+        recommended_actions.append(
+            RecommendedAction(
+                priority="medium",
+                title="Review warning quality checks",
+                detail=f"{summary.warning_quality_checks} warning checks may need follow-up.",
+            )
+        )
+
+    service_status = "healthy"
+    if summary.failed_quality_checks or latest_failed_pipelines or stale_pipeline_runs:
+        service_status = "attention_required"
+    elif summary.warning_quality_checks:
+        service_status = "degraded"
+
+    return OperationsOverview(
+        generated_at=generated_at,
+        service_status=service_status,
+        summary=summary,
+        pipeline_health=pipeline_health,
+        quality_checks=quality_checks,
+        stale_pipeline_runs=stale_pipeline_runs,
+        recommended_actions=recommended_actions,
+    )

tests/test_pipeline_runs.py

@@ -113,6 +113,31 @@ def test_metrics_summary_counts_runs_and_checks(client: TestClient) -> None:
     assert payload["warning_quality_checks"] == 0
 
 
+def test_operations_overview_returns_operator_snapshot(
+    client: TestClient,
+    db_session: Session,
+) -> None:
+    seed_sample_data(db_session)
+
+    response = client.get("/api/v1/metrics/operations-overview?stale_after_minutes=10")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["service_status"] == "attention_required"
+    assert payload["summary"]["total_runs"] == 3
+    assert [rollup["name"] for rollup in payload["pipeline_health"]] == [
+        "inventory_snapshot",
+        "orders_daily_load",
+    ]
+    assert payload["quality_checks"][0]["severity"] == "critical"
+    assert [run["name"] for run in payload["stale_pipeline_runs"]] == ["inventory_snapshot"]
+    assert [action["priority"] for action in payload["recommended_actions"]] == [
+        "critical",
+        "high",
+        "medium",
+    ]
+
+
 def test_pipeline_health_rollups_group_runs_by_pipeline_name(
     client: TestClient,
     db_session: Session,