Trigger DAGs and expose datasets

The last two routes round out the control plane. /pipeline/trigger asks Airflow to run the batch DAG now, with an idempotent run id. /datasets lists the warehouse tables a downstream app is allowed to read. Both routes degrade gracefully: if the warehouse is down, /datasets returns an empty list with a 200, not a 500.

router.py
python
from uuid import uuid4
import httpx

@router.post('/pipeline/trigger', response_model=PipelineTriggerResponse)
async def trigger(req: BatchRequest) -> PipelineTriggerResponse:
    run_id = str(uuid4())
    if req.trigger_airflow and service.airflow.base_url:
        try:
            async with httpx.AsyncClient(
                timeout=service.airflow.request_timeout_seconds,
                auth=(service.airflow.username, service.airflow.password),
            ) as client:
                await client.post(
                    f'{service.airflow.base_url}/api/v1/dags/{service.airflow.batch_dag_id}/dagRuns',
                    json={'dag_run_id': run_id, 'conf': req.model_dump()},
                )
        except Exception:
            pass
    return PipelineTriggerResponse(status='accepted', dag_id=service.airflow.batch_dag_id, run_id=run_id)

The trigger route generates a UUID run id and calls the Airflow REST API. When Airflow is unreachable we still return accepted with the run id, so the caller has a trace id to follow up with even though Airflow received no request yet.

Accepted is not a promise of success. It is a promise of receipt. The caller now owns a run_id they can poll. If Airflow was down, the DAG run will not exist, and the caller will find out on the next list call. This removes the temptation to retry the trigger in a loop and create duplicate runs.

router.py
python
@router.get('/datasets', response_model=list[DatasetSummary])
async def list_datasets() -> list[DatasetSummary]:
    if not service.db.postgres:
        return []
    try:
        return await service.list_datasets()
    except Exception as exc:
        log.warning('datasets.unavailable', error=str(exc)[:200])
        return []

List datasets returns what the warehouse exposes. If Postgres is down, it returns an empty list and logs the reason, so the dashboard shows "no datasets available" instead of a red toast with a 500.

Two options. One, require the caller to supply the run_id so two identical clicks produce the same run_id and Airflow rejects the second as duplicate. Two, check if a run with the same logical run_id is active before you create another. The first is cheap and moves the dedup responsibility to the caller. The second is more complex but friendlier for anonymous UIs.

Validation checklist: Verify your control plane

Loading practice…

Quiz: Quiz

Loading practice…

Checkpoint: Final checkpoint: you own a data platform

Loading practice…