Pipelines

A pipeline takes data from a source, applies a transform, and loads the result into a Pact object. Pipelines are managed at /v1/etl/pipelines (backed by the etl_pipelines + etl_runs tables, alembic 0098_etl_pipelines) and listed in the app under Data → Pipelines.

Each pipeline row holds its definition plus a rolled-up snapshot of its most recent run (last_run_at, last_run_status). Every execution attempt is recorded as its own etl_runs row with full timing, record counts, and a log payload — so the run history is real even where the execution is simulated.

Execution is an inline simulation today

Pipeline CRUD and the run-history table are fully real and tenant-isolated. The execute step runs inline as a mock — _run_pipeline_mock() in api/routes/etl.py inserts a running row, generates records_in / records_out / records_failed plus a synthetic log, and writes the terminal status back. No real transform is applied and no records land in the target object yet. The route contract is stable, so a production dispatcher can replace the mock without changing the API.

Destinations

A pipeline's destination is the Pact object it loads into, validated against a fixed allowlist (ALLOWED_DESTINATIONS):

companies · contacts · events · opportunities · tags · tracking_events · custom

Pipelines carry a schedule field (default manual) and a user-settable status of active or paused. The schedule value is stored, but there is no background scheduler dispatching scheduled runs yet — runs are triggered explicitly via the API or the inline trigger on each pipeline card.

Endpoints

code

POST   /v1/etl/pipelines               create
GET    /v1/etl/pipelines               list
GET    /v1/etl/pipelines/{id}          fetch one
PATCH  /v1/etl/pipelines/{id}          update
DELETE /v1/etl/pipelines/{id}          delete + cascade runs
GET    /v1/etl/pipelines/{id}/runs     paginated run history
POST   /v1/etl/pipelines/{id}/runs     enqueue + execute (inline mock)

bash

curl -X POST https://api.pact.place/v1/etl/pipelines \
  -H "Authorization: Bearer $PACT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"name":"Salesforce → Contacts","destination":"contacts","transform":{}}'

A pipeline optionally references a source_id; the transform is stored as transform_json. Deleting a pipeline cascades its etl_runs rows.

Pipeline detailRun history, trigger controls, and per-run record counts.Data sourcesThe upstream connections pipelines read from.Data vaultThe outbound counterpart — export Pact data to your warehouse.