From c7eef7a72a2f8475dec5fc95369f4e21e2e15420 Mon Sep 17 00:00:00 2001 From: Bmowville Date: Sun, 7 Jun 2026 00:36:28 -0400 Subject: [PATCH 1/2] Add operations overview endpoint --- .env.example | 3 ++ .github/workflows/ci.yml | 36 +++++++++++++++++ .gitignore | 14 +++++++ README.md | 13 ++++++ app/api/pipeline_runs.py | 10 +++++ app/schemas/pipeline.py | 16 ++++++++ app/services/pipeline_runs.py | 76 +++++++++++++++++++++++++++++++++++ tests/test_pipeline_runs.py | 25 ++++++++++++ 8 files changed, 193 insertions(+) diff --git a/.env.example b/.env.example index 6197c36..afca4dc 100644 --- a/.env.example +++ b/.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 \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e5b2b69..5ce9653 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,5 +1,41 @@ name: CI +on: + pull_request: + push: + branches: + - main + +jobs: + quality: + runs-on: ubuntu-latest + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[dev]" + + - name: Lint + run: ruff check . + + - name: Type check + run: mypy app + + - name: Run migrations + run: alembic upgrade head + + - name: Test + run: pytestname: CI + on: push: branches: ["main"] diff --git a/.gitignore b/.gitignore index 11e14b4..51e5085 100644 --- a/.gitignore +++ b/.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 diff --git a/README.md b/README.md index 8b06c24..e65ae44 100644 --- a/README.md +++ b/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. diff --git a/app/api/pipeline_runs.py b/app/api/pipeline_runs.py index 6d060ed..0865b0e 100644 --- a/app/api/pipeline_runs.py +++ b/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) diff --git a/app/schemas/pipeline.py b/app/schemas/pipeline.py index c26b5fd..c1c610f 100644 --- a/app/schemas/pipeline.py +++ b/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 diff --git a/app/services/pipeline_runs.py b/app/services/pipeline_runs.py index edad659..a61cbab 100644 --- a/app/services/pipeline_runs.py +++ b/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, + ) diff --git a/tests/test_pipeline_runs.py b/tests/test_pipeline_runs.py index b92ed35..304f949 100644 --- a/tests/test_pipeline_runs.py +++ b/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, From 97be515f6de7bc15b544da50e5f3f8ef3d093dd8 Mon Sep 17 00:00:00 2001 From: Bmowville Date: Sun, 7 Jun 2026 00:38:16 -0400 Subject: [PATCH 2/2] Fix CI workflow syntax --- .github/workflows/ci.yml | 39 +++++---------------------------------- 1 file changed, 5 insertions(+), 34 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5ce9653..0953e96 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,10 +1,13 @@ name: CI on: - pull_request: push: branches: - main + pull_request: + +permissions: + contents: read jobs: quality: @@ -34,36 +37,4 @@ jobs: run: alembic upgrade head - name: Test - run: pytestname: CI - -on: - push: - branches: ["main"] - pull_request: - -permissions: - contents: read - -jobs: - quality: - runs-on: ubuntu-latest - steps: - - name: Checkout - uses: actions/checkout@v4 - - name: Setup Python - uses: actions/setup-python@v5 - with: - python-version: "3.11" - cache: pip - - name: Install - run: | - python -m pip install --upgrade pip - python -m pip install -e ".[dev]" - - name: Ruff - run: ruff check . - - name: Mypy - run: mypy app - - name: Migrations - run: alembic upgrade head - - name: Tests - run: pytest --cov=app --cov-report=term-missing \ No newline at end of file + run: pytest \ No newline at end of file