Add webhook alerts for pipeline incidents

.env.example

@@ -3,4 +3,10 @@ ENVIRONMENT=local
 DATABASE_URL=sqlite:///./dataops_observability.db
 API_PREFIX=/api/v1
 # Optional comma-separated keys. When set, write/ingestion endpoints require X-DataOps-API-Key.
-INGESTION_API_KEYS=
+INGESTION_API_KEYS=
+PUBLIC_BASE_URL=http://127.0.0.1:8000
+# Optional comma-separated alert receivers. Failed/canceled runs and warning/failed checks are posted here.
+ALERT_WEBHOOK_URLS=
+# Optional shared secret sent as X-DataOps-Webhook-Secret to alert receivers.
+ALERT_WEBHOOK_SECRET=
+ALERT_WEBHOOK_TIMEOUT_SECONDS=5

README.md

@@ -15,6 +15,7 @@ Most pipeline monitoring tools stop at raw events, logs, or generic charts. This
 - a small dashboard that makes the API useful immediately after seeding sample data
 - integration-friendly endpoints for Airflow, dbt, cron jobs, GitHub Actions, or custom ETL scripts
 - optional API-key protection for ingestion writes when the service is exposed outside a local machine
+- webhook alerts for failed pipeline runs and quality checks that need operator attention
 - a codebase small enough for teams to adapt instead of adopting a heavy platform
 
 See [docs/integrations.md](docs/integrations.md) for copy-paste examples that report pipeline events from Python jobs, GitHub Actions, Airflow, and dbt.
@@ -73,6 +74,8 @@ Full setup notes live in [docs/integrations.md](docs/integrations.md).
 
 Set `INGESTION_API_KEYS` to require external reporters to send `X-DataOps-API-Key` on write requests. Read-only dashboard and metrics endpoints stay open so operators can inspect service health without sharing ingestion credentials.
 
+Set `ALERT_WEBHOOK_URLS` to send operational alerts to Slack-compatible bridges, incident tooling, or custom automation when runs fail/cancel or quality checks warn/fail.
+
 ## Run Quality Gates
 
 ```powershell
@@ -168,6 +171,10 @@ Settings are loaded from environment variables.
 | `DATABASE_URL` | `sqlite:///./dataops_observability.db` | SQLAlchemy database URL |
 | `API_PREFIX` | `/api/v1` | Versioned API prefix |
 | `INGESTION_API_KEYS` | empty | Comma-separated keys accepted by write/ingestion endpoints. When empty, local writes are open. |
+| `PUBLIC_BASE_URL` | `http://127.0.0.1:8000` | Base URL used in generated dashboard and API links. |
+| `ALERT_WEBHOOK_URLS` | empty | Comma-separated webhook URLs that receive operational alerts. |
+| `ALERT_WEBHOOK_SECRET` | empty | Optional shared secret sent as `X-DataOps-Webhook-Secret` on alert deliveries. |
+| `ALERT_WEBHOOK_TIMEOUT_SECONDS` | `5` | Timeout for each outbound webhook delivery. |
 
 When `INGESTION_API_KEYS` is set, these write endpoints require `X-DataOps-API-Key`:
 
@@ -177,6 +184,27 @@ When `INGESTION_API_KEYS` is set, these write endpoints require `X-DataOps-API-K
 
 Use different keys during rotation by setting a comma-separated value such as `INGESTION_API_KEYS=old-key,new-key`. Store real keys in your deployment secret manager, CI secrets, or local `.env` file rather than committing them.
 
+## Webhook Alerts
+
+When `ALERT_WEBHOOK_URLS` is configured, the API sends background webhook alerts for events that normally need human attention:
+
+- pipeline run status changes to `failed`
+- pipeline run status changes to `canceled`
+- quality check status is `failed`
+- quality check status is `warning`
+
+Successful runs and passed quality checks do not send alerts. This keeps notification volume focused on work that needs follow-up.
+
+Example configuration:
+
+```powershell
+$env:PUBLIC_BASE_URL = "https://dataops.example.com"
+$env:ALERT_WEBHOOK_URLS = "https://hooks.example.com/dataops,https://backup.example.com/dataops"
+$env:ALERT_WEBHOOK_SECRET = "shared-secret"
+```
+
+Alert payloads include `event_type`, `severity`, `message`, the affected pipeline run, optional quality-check details, and links back to the dashboard, run detail endpoint, and timeline endpoint.
+
 ## Notes
 
 SQLite is the default for local development. The SQLAlchemy and Alembic setup is structured so the service can be moved to Postgres by changing `DATABASE_URL` and adding the relevant deployment configuration.

app/api/pipeline_runs.py

@@ -1,8 +1,9 @@
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
 from sqlalchemy.orm import Session
 
+from app.core.config import Settings, get_settings
 from app.core.security import require_ingestion_api_key
 from app.db.session import get_db
 from app.models.pipeline import PipelineRun
@@ -35,6 +36,7 @@
     list_quality_checks,
     update_pipeline_run,
 )
+from app.services.webhook_alerts import queue_pipeline_run_alert, queue_quality_check_alert
 
 router = APIRouter(tags=["pipeline runs"])
 
@@ -54,9 +56,13 @@ def _get_run_or_404(db: Session, run_id: int) -> PipelineRun:
 )
 def create_run(
     payload: PipelineRunCreate,
+    background_tasks: BackgroundTasks,
     db: Annotated[Session, Depends(get_db)],
+    settings: Annotated[Settings, Depends(get_settings)],
 ) -> PipelineRun:
-    return create_pipeline_run(db, payload)
+    run = create_pipeline_run(db, payload)
+    queue_pipeline_run_alert(background_tasks, settings, run)
+    return run
 
 
 @router.get("/pipeline-runs", response_model=list[PipelineRunRead])
@@ -101,10 +107,15 @@ def read_run_timeline(
 def patch_run(
     run_id: int,
     payload: PipelineRunUpdate,
+    background_tasks: BackgroundTasks,
     db: Annotated[Session, Depends(get_db)],
+    settings: Annotated[Settings, Depends(get_settings)],
 ) -> PipelineRun:
     run = _get_run_or_404(db, run_id)
-    return update_pipeline_run(db, run, payload)
+    updated_run = update_pipeline_run(db, run, payload)
+    if payload.status in {PipelineRunStatus.failed, PipelineRunStatus.canceled}:
+        queue_pipeline_run_alert(background_tasks, settings, updated_run)
+    return updated_run
 
 
 @router.post(
@@ -116,10 +127,14 @@ def patch_run(
 def create_check(
     run_id: int,
     payload: QualityCheckCreate,
+    background_tasks: BackgroundTasks,
     db: Annotated[Session, Depends(get_db)],
+    settings: Annotated[Settings, Depends(get_settings)],
 ) -> QualityCheckRead:
     run = _get_run_or_404(db, run_id)
-    return QualityCheckRead.model_validate(create_quality_check(db, run, payload))
+    check = create_quality_check(db, run, payload)
+    queue_quality_check_alert(background_tasks, settings, run, check)
+    return QualityCheckRead.model_validate(check)
 
 
 @router.get("/pipeline-runs/{run_id}/quality-checks", response_model=list[QualityCheckRead])

app/core/config.py

@@ -13,6 +13,23 @@ class Settings(BaseSettings):
         default="",
         description="Comma-separated API keys accepted for write/ingestion endpoints.",
     )
+    public_base_url: str = Field(
+        default="http://127.0.0.1:8000",
+        description="Base URL used in generated dashboard and API links.",
+    )
+    alert_webhook_urls: str = Field(
+        default="",
+        description="Comma-separated webhook URLs that receive operational alerts.",
+    )
+    alert_webhook_secret: str = Field(
+        default="",
+        description="Optional shared secret sent with webhook alert deliveries.",
+    )
+    alert_webhook_timeout_seconds: float = Field(
+        default=5.0,
+        gt=0,
+        description="Timeout in seconds for outbound webhook alert deliveries.",
+    )
 
     model_config = SettingsConfigDict(env_file=".env", extra="ignore")
 

app/services/webhook_alerts.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import json
+import logging
+from datetime import UTC, datetime
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+from fastapi import BackgroundTasks
+
+from app.core.config import Settings
+from app.models.pipeline import PipelineRun, QualityCheck
+from app.schemas.pipeline import PipelineRunStatus, QualityCheckStatus
+
+WEBHOOK_SECRET_HEADER = "X-DataOps-Webhook-Secret"
+WEBHOOK_USER_AGENT = "dataops-observability-api/0.1"
+ACTIONABLE_RUN_STATUSES = {
+    PipelineRunStatus.failed.value,
+    PipelineRunStatus.canceled.value,
+}
+ACTIONABLE_CHECK_STATUSES = {
+    QualityCheckStatus.failed.value,
+    QualityCheckStatus.warning.value,
+}
+
+AlertPayload = dict[str, object]
+
+logger = logging.getLogger(__name__)
+
+
+def get_configured_alert_webhook_urls(settings: Settings) -> tuple[str, ...]:
+    return tuple(
+        webhook_url.strip()
+        for webhook_url in settings.alert_webhook_urls.split(",")
+        if webhook_url.strip()
+    )
+
+
+def queue_pipeline_run_alert(
+    background_tasks: BackgroundTasks,
+    settings: Settings,
+    run: PipelineRun,
+) -> None:
+    if run.status not in ACTIONABLE_RUN_STATUSES:
+        return
+
+    queue_webhook_alert(
+        background_tasks,
+        settings,
+        build_pipeline_run_alert_payload(run, settings),
+    )
+
+
+def queue_quality_check_alert(
+    background_tasks: BackgroundTasks,
+    settings: Settings,
+    run: PipelineRun,
+    check: QualityCheck,
+) -> None:
+    if check.status not in ACTIONABLE_CHECK_STATUSES:
+        return
+
+    queue_webhook_alert(
+        background_tasks,
+        settings,
+        build_quality_check_alert_payload(run, check, settings),
+    )
+
+
+def queue_webhook_alert(
+    background_tasks: BackgroundTasks,
+    settings: Settings,
+    payload: AlertPayload,
+) -> None:
+    webhook_urls = get_configured_alert_webhook_urls(settings)
+    if not webhook_urls:
+        return
+
+    background_tasks.add_task(
+        send_webhook_alerts,
+        webhook_urls,
+        payload,
+        settings.alert_webhook_secret,
+        settings.alert_webhook_timeout_seconds,
+    )
+
+
+def build_pipeline_run_alert_payload(run: PipelineRun, settings: Settings) -> AlertPayload:
+    event_type = f"pipeline_run_{run.status}"
+    severity = "critical" if run.status == PipelineRunStatus.failed.value else "warning"
+    return {
+        "event_type": event_type,
+        "severity": severity,
+        "message": f"Pipeline run {run.name} is {run.status}.",
+        "occurred_at": datetime.now(UTC).isoformat(),
+        "pipeline_run": serialize_pipeline_run(run),
+        "links": build_alert_links(run, settings),
+    }
+
+
+def build_quality_check_alert_payload(
+    run: PipelineRun,
+    check: QualityCheck,
+    settings: Settings,
+) -> AlertPayload:
+    event_type = f"quality_check_{check.status}"
+    severity = check.severity
+    return {
+        "event_type": event_type,
+        "severity": severity,
+        "message": f"Quality check {check.check_name} is {check.status} for {run.name}.",
+        "occurred_at": datetime.now(UTC).isoformat(),
+        "pipeline_run": serialize_pipeline_run(run),
+        "quality_check": serialize_quality_check(check),
+        "links": build_alert_links(run, settings),
+    }
+
+
+def serialize_pipeline_run(run: PipelineRun) -> dict[str, object]:
+    return {
+        "id": run.id,
+        "name": run.name,
+        "source_system": run.source_system,
+        "status": run.status,
+        "records_processed": run.records_processed,
+        "started_at": format_datetime(run.started_at),
+        "finished_at": format_datetime(run.finished_at),
+        "error_message": run.error_message,
+        "created_at": run.created_at.isoformat(),
+        "updated_at": run.updated_at.isoformat(),
+    }
+
+
+def serialize_quality_check(check: QualityCheck) -> dict[str, object]:
+    return {
+        "id": check.id,
+        "pipeline_run_id": check.pipeline_run_id,
+        "check_name": check.check_name,
+        "status": check.status,
+        "severity": check.severity,
+        "expected_value": check.expected_value,
+        "observed_value": check.observed_value,
+        "details": check.details,
+        "created_at": check.created_at.isoformat(),
+    }
+
+
+def build_alert_links(run: PipelineRun, settings: Settings) -> dict[str, str]:
+    base_url = settings.public_base_url.rstrip("/")
+    return {
+        "dashboard": f"{base_url}/dashboard",
+        "pipeline_run": f"{base_url}{settings.api_prefix}/pipeline-runs/{run.id}",
+        "timeline": f"{base_url}{settings.api_prefix}/pipeline-runs/{run.id}/timeline",
+    }
+
+
+def format_datetime(value: datetime | None) -> str | None:
+    if value is None:
+        return None
+    return value.isoformat()
+
+
+def send_webhook_alerts(
+    webhook_urls: tuple[str, ...],
+    payload: AlertPayload,
+    shared_secret: str,
+    timeout_seconds: float,
+) -> None:
+    for webhook_url in webhook_urls:
+        send_webhook_alert(webhook_url, payload, shared_secret, timeout_seconds)
+
+
+def send_webhook_alert(
+    webhook_url: str,
+    payload: AlertPayload,
+    shared_secret: str,
+    timeout_seconds: float,
+) -> None:
+    data = json.dumps(payload).encode("utf-8")
+    headers = {
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+        "User-Agent": WEBHOOK_USER_AGENT,
+    }
+    if shared_secret:
+        headers[WEBHOOK_SECRET_HEADER] = shared_secret
+
+    request = Request(
+        url=webhook_url,
+        data=data,
+        headers=headers,
+        method="POST",
+    )
+    try:
+        with urlopen(request, timeout=timeout_seconds):
+            return
+    except (HTTPError, URLError, TimeoutError, OSError) as error:
+        logger.warning("Webhook alert delivery failed for %s: %s", webhook_url, error)