DataQ — System Architecture¶
Keep these diagrams in sync with the code. When a new component, datasource, integration, or DB table is added, update the relevant diagram in the same PR.
☁️ These diagrams are drawn using Azure's component names (Container Apps, Key Vault, App Insights, Azure AD) because that's the concrete deployment they trace through. DataQ runs as an equally-live peer deployment on AWS (ECS Fargate, Secrets Manager, CloudWatch+X-Ray, Cognito) behind the exact same seams — see deployment parity for the side-by-side. Neither cloud is primary; read "Key Vault" / "App Insights" / "Azure AD" below as "the secret store" / "the observability backend" / "the OIDC authority" and substitute the AWS equivalent where relevant.
%%{init: {'flowchart': {'curve': 'linear'}}}%%
flowchart LR
subgraph orch["⚙️ Orchestration — monitor + trigger only"]
ADF["Azure Data Factory"]
AF["Apache Airflow"]
DBT["dbt"]
end
subgraph clients["🌐 Clients"]
AICli["AI clients · Claude/Copilot/Cursor"]
Web["Web UI · React SPA"]
end
subgraph platform["🟦 DataQ Platform · Azure Container Apps"]
Frontend["Frontend · nginx + React SPA<br/>sole public ingress · runtime OIDC config"]
API["FastAPI · internal ingress<br/>REST /api/v1 + /mcp · OIDC JWT"]
Worker["Celery worker<br/>GX Core execution"]
Beat["Celery beat · own process<br/>schedules only, never executes"]
Infra["PostgreSQL · Redis · Key Vault · App Insights<br/>suites·runs·results·pipeline_runs · queue · secrets · traces"]
end
subgraph alerts["🔔 Alerts · ResultPublisher seam"]
Teams["Teams · Slack"]
Email["Email · SMTP"]
end
subgraph ds["📊 Datasources — GX checks execute here"]
SF["Snowflake · DEV/QA/UAT"]
Files["ADLS Gen2 · S3 · GCS · flat files"]
UC["Unity Catalog · Databricks"]
Iceberg["Apache Iceberg · native pyiceberg read<br/>object storage + catalog"]
PGDS["PostgreSQL · MySQL / MariaDB · Trino · SQL Server · Athena · Redshift · BigQuery<br/>generic SQL base, pushdown"]
end
Web -->|HTTPS| Frontend
AICli -->|MCP · HTTP| Frontend
orch -->|"webhook / HMAC callback<br/>+ 10-min REST poll fallback"| Frontend
Frontend -->|"same-origin proxy<br/>/api + /mcp + /healthz"| API
API -->|enqueue run| Worker
API -.-> Infra
Worker -.-> Infra
Beat -.->|schedule tick, enqueue| Infra
Worker -->|run outcome| alerts
Worker -->|GX checks| ds
classDef hub fill:#E6F1FB,stroke:#185FA5,color:#0C449C
classDef src fill:#E1F5EE,stroke:#0F6E56,color:#085041
classDef integ fill:#FAEEDA,stroke:#854F0B,color:#633806
classDef notify fill:#FAECE7,stroke:#993C1D,color:#712B13
classDef client fill:#F1EFE8,stroke:#5F5E5A,color:#444441
class API,Worker,Beat,Infra,Frontend hub
class SF,Files,UC,Iceberg src
class ADF,AF,DBT integ
class Teams,Email notify
class Web,AICli client
style platform fill:#F4F9FE,stroke:#185FA5
style ds fill:#F0FBF7,stroke:#0F6E56
style orch fill:#FDF7EC,stroke:#854F0B
style alerts fill:#FDF3EF,stroke:#993C1D
style clients fill:#F7F6F2,stroke:#5F5E5A
linkStyle default stroke:#5B6B7B,stroke-width:1.5pxThe flow reads left → right: inputs (Clients, Orchestration) drive the DataQ platform in the centre, which acts on its targets on the right (runs GX checks against datasources, publishes outcomes to alert channels).
Legend¶
| Colour | Group |
|---|---|
| Grey | Clients — browser web UI + AI clients (over MCP) |
| Orange | Orchestration — ADF · Airflow · dbt (monitor + trigger only, never datasources) |
| Blue | DataQ platform — Frontend (nginx SPA) · FastAPI · Celery worker · Celery beat (own process) · PostgreSQL · Redis · Key Vault · App Insights |
| Green | Datasources — GX checks run against these |
| Red | Alert channels — Teams · Slack · Email (the ResultPublisher seam) |
Data model (ER diagram)¶
Source of truth:
backend/app/db/models.py(32 tables). Update this diagram in the same PR as any model/migration change.
erDiagram
users {
uuid id PK
string aad_object_id UK "nullable — NULL for non-AAD (email-OTP) identities"
string oidc_issuer "issuer that vouched for aad_object_id (provider-neutral OIDC)"
string email UK "unique on lower(email) — the cross-authenticator identity key"
string display_name
bool display_name_override "user-set name survives IdP re-sync"
string role "admin | member | viewer — CHECK-constrained, ADR 0033"
timestamptz last_seen_at
}
workspace_members {
uuid id PK
string email UK "unique on lower(email) - who is admitted to the workspace"
string initial_role "seeds the users row on first sign-in only - ADR 0043"
string source "admin | auto_import - auto_import rows await review"
uuid invited_by FK "SET NULL - the membership outlives the inviter"
timestamptz created_at
}
api_keys {
uuid id PK
uuid user_id FK "CASCADE"
string name "label, e.g. ci-smoke"
string key_prefix "dq_live_ + 4 — identifies without revealing"
string key_hash UK "sha256 hex; plaintext never stored"
timestamptz expires_at "mandatory expiry"
timestamptz revoked_at "soft revocation"
timestamptz last_used_at "throttled telemetry"
}
sessions {
uuid id PK
uuid user_id FK "CASCADE"
string token_hash UK "sha256 hex of the dq_sess_ cookie token"
timestamptz expires_at "fixed horizon — no refresh pair"
timestamptz revoked_at "logout"
}
otp_codes {
uuid id PK
string email "normalized (strip+lower); deliberately NOT FK'd to users"
string code_hash "sha256 hex of the 6-digit code"
timestamptz expires_at "10-minute TTL"
timestamptz consumed_at "redeemed OR superseded by a re-request"
int attempts "atomically incremented before each compare; capped"
}
connections {
uuid id PK
string name "unique per env"
string type "snowflake / adls_gen2 / s3 / gcs / unity_catalog / iceberg / postgres / mysql / trino / mssql / athena / redshift / bigquery / adf / airflow / dbt"
string env "dev / qa / uat / prod"
jsonb config "non-secret datasource config"
string secret_ref "SecretStore key, never the credential"
uuid created_by FK
timestamptz last_polled_at "orchestration poll health: + last_poll_error, consecutive_poll_failures, health_alerted_at"
timestamptz credential_expires_at "readable credential lifetime: + credential_expiry_checked_at — NULL = unknown, never 'does not expire'"
timestamptz lineage_last_refresh_at "lineage pull state (ADR 0034): + lineage_watermark, lineage_last_tier, lineage_degraded_reason, lineage_last_error"
timestamptz inventory_sync_last_attempted_at "inventory sync state (ADR 0040): + inventory_sync_last_error, _failing_since, _last_table_count, _zero_since"
}
connection_versions {
uuid id PK
uuid connection_id FK
int version_no "per-connection sequence"
string name
string type
string env
jsonb config
uuid changed_by FK "SET NULL"
}
assets {
uuid id PK
string namespace "OpenLineage dataset namespace (ADR 0034)"
string name "OpenLineage dataset name — (namespace,name) unique"
string env "metadata, not identity"
uuid connection_id FK "provenance hint (SET NULL)"
uuid owner_user_id FK "incident-routing hop (SET NULL — reserved for future routing use)"
string description "workspace-Admin-set (ADR 0034 §4)"
jsonb column_tags "warehouse governance classifications, cached per read"
timestamptz column_tags_refreshed_at "'never looked' vs 'looked, found none'"
timestamptz detected_classifications_refreshed_at "last profile that fed the profiler classification source (ADR 0049)"
bool auto_coverage_excluded "left out of its connection's automatic coverage (ADR 0047)"
jsonb auto_coverage_config "this table's sensitivity for its automatic baselines; NULL = workspace setting"
timestamptz first_seen
timestamptz last_seen
}
asset_column_classifications {
uuid id PK
uuid asset_id FK "CASCADE — a classification is a fact about this asset (ADR 0049)"
string column_name "case-folded unique per (asset, source)"
string source "profiler today; the governance-catalog pull next"
string verdict "always sensitive — additive only, never a clearance"
string detector "email / encoded"
jsonb evidence "detector counts, never a value"
timestamptz first_detected_at
timestamptz last_detected_at "keeps moving after a dismissal, so the page can say 'still detected'"
timestamptz dismissed_at "cleared by a workspace Admin, never by a re-profile"
uuid dismissed_by FK "SET NULL on erasure"
timestamptz created_at
}
lineage_edges {
uuid id PK
uuid upstream_asset_id FK "CASCADE (ADR 0034)"
uuid downstream_asset_id FK "CASCADE"
string source "lineage source, e.g. 'dbt'"
uuid connection_id FK "CASCADE — refreshing conn (provenance + prune scope); (up,down,source,connection_id) unique"
jsonb columns "column-level mapping when the source provides one (none_as_null)"
timestamptz first_seen
timestamptz last_seen
}
suites {
uuid id PK
string name
string description
uuid connection_id FK
uuid asset_id FK "resolved target asset (SET NULL, fail-soft)"
jsonb target "table / path / UC name the checks run against"
jsonb column_policy "failing-sample redaction policy"
string origin "user / auto — auto = owned by the coverage loop, one per asset (ADR 0047)"
jsonb auto_state "coverage-loop bookkeeping: deleted checks, recorded gaps"
uuid contract_id FK "the data contract it enforces (ADR 0052; SET NULL)"
string contract_object "the ODCS schema object — set and cleared with contract_id"
uuid created_by FK
}
contracts {
uuid id PK
string odcs_id "the document's own id — identity with version (ADR 0052)"
string version
string name "display only"
string api_version "the ODCS apiVersion the document declared"
jsonb spec "the ODCS document, verbatim"
jsonb unmapped "clauses no check enforces, with the reason"
uuid created_by FK
}
checks {
uuid id PK
uuid suite_id FK
string name
string kind "expectation / freshness / volume / schema_drift / anomaly / comparison (ADR 0012/0015)"
string expectation_type
string origin "user / auto / suggestion (ADR 0047) / contract (ADR 0052)"
jsonb contract_ref "the contract clause a contract-origin check enforces"
string engine "gx / dmf / dqx / dataplex — per-check evaluator, default gx (ADR 0036)"
string dimension "DQ dimension (ADR 0038) — nullable; NULL = unclassified"
uuid source_connection_id FK "comparison baseline datasource (ADR 0015; RESTRICT) — set iff kind='comparison'"
numeric warn_threshold
numeric fail_threshold
numeric critical_threshold
jsonb config "GX expectation kwargs"
timestamptz alert_snoozed_until
bool enabled "false = kept with its history, left out of runs"
}
monitor_baselines {
uuid id PK
uuid check_id FK "UNIQUE - one current baseline"
string kind "denormalized from check"
jsonb baseline "kind-shaped: schema snapshot / anomaly params"
timestamptz captured_at
uuid captured_by FK "NULL = run-captured"
}
check_suggestions {
uuid id PK
uuid suite_id FK "CASCADE — the queue belongs to its automatic suite (ADR 0047)"
string source "profile / llm"
string status "pending / accepted / rejected — (suite_id, fingerprint) unique, so a rejection is remembered"
string fingerprint "sha256 of expectation type + canonical config"
string expectation_type
jsonb config
uuid check_id FK "SET NULL — the check created on accept"
uuid decided_by FK "SET NULL"
timestamptz decided_at
}
check_versions {
uuid id PK
uuid check_id FK
int version_no "per-check sequence"
string name
string kind
string expectation_type
string dimension "DQ dimension (ADR 0038) — nullable; NULL = unclassified"
uuid source_connection_id "snapshot — plain UUID, deliberately no FK"
jsonb config
numeric warn_threshold
numeric fail_threshold
numeric critical_threshold
bool enabled
uuid changed_by FK "SET NULL"
}
audit_events {
uuid id PK
timestamptz occurred_at
string action_class "config (ADR 0041 phase 1) / access (phase 2)"
string action "check.update / share.grant / connection.reauth / …"
string entity_type
uuid entity_id "NO FK — the audit row must outlive the entity"
uuid actor_user_id FK "SET NULL"
string actor_kind "user / pat / webhook — deliberately no 'system'"
string actor_label "denormalized, survives the SET NULL"
jsonb before
jsonb after
string request_id
string prev_hash "hash chain (ADR 0041 §9)"
string row_hash "NULL = written before the chain shipped, not backfilled"
}
audit_chain_state {
int id PK "singleton — always 1"
string head_hash "locked via FOR UPDATE at commit time"
uuid head_event_id "NO FK — see audit_events.entity_id"
timestamptz updated_at
}
audit_chain_checkpoints {
uuid id PK
timestamptz created_at
timestamptz cutoff "the retention cutoff this purge ran against"
int deleted_count
uuid last_deleted_event_id "NO FK — about to be deleted"
string last_deleted_row_hash
uuid first_surviving_event_id "NO FK — nullable if the purge deleted everything"
boolean anchored "whether the external anchor publish succeeded"
}
runs {
uuid id PK
uuid suite_id FK
uuid asset_id FK "stamped at dispatch (SET NULL) — targets change; history shouldn't rewrite"
string status "queued / running / succeeded / failed / cancelled"
string triggered_by "manual / schedule / provider:pipeline:run_id"
string celery_task_id
string failure_reason "operational failure classification"
timestamptz started_at
timestamptz finished_at
}
results {
uuid id PK
uuid run_id FK
uuid check_id FK
string status "pass warn fail critical + skip error"
numeric metric_value "SQL-aggregatable scalar (ADR 0012)"
int duration_ms
jsonb observed_value
jsonb expected_value
jsonb sample_failures "redacted failing rows"
timestamptz sample_failures_purged_at
jsonb sampling "scale-aware read strategy + counts — NULL = full read"
}
shares {
uuid id PK
uuid suite_id FK
uuid user_id FK
string permission "view / edit (admin schema-legal, never granted)"
}
tags {
uuid id PK
string name "unique; normalized lower-case label (a key:value form fits inside it)"
}
suite_tags {
uuid suite_id PK,FK "CASCADE"
uuid tag_id PK,FK "CASCADE"
}
asset_tags {
uuid asset_id PK,FK "CASCADE"
uuid tag_id PK,FK "CASCADE"
}
connection_tags {
uuid connection_id PK,FK "CASCADE"
uuid tag_id PK,FK "CASCADE"
}
check_tags {
uuid check_id PK,FK "CASCADE"
uuid tag_id PK,FK "CASCADE"
}
pipeline_runs {
uuid id PK
string provider "adf / airflow / dbt"
uuid connection_id FK
string provider_run_id "unique per provider"
string pipeline_or_dag_id
string env
string status
timestamptz started_at
timestamptz finished_at
string failure_reason
timestamptz last_updated_at "staleness guard for the polling upsert"
}
trigger_bindings {
uuid id PK
string provider
string pipeline_or_dag_id
string env
uuid suite_id FK
bool enabled
}
schedules {
uuid id PK
uuid suite_id FK
string cron "5-field, evaluated in timezone"
string timezone "IANA name, default UTC"
bool enabled
timestamptz next_run_at "precomputed next fire"
timestamptz last_run_at
uuid created_by FK
}
suite_notifications {
uuid id PK
uuid suite_id FK "unique - one row per suite"
bool enabled
string alert_on "fail / warn / always"
string webhook_secret_ref "per-suite Teams webhook, SecretStore key"
string slack_webhook_secret_ref "per-suite Slack webhook ref (NULL → workspace webhook)"
string email_recipients "comma-separated addresses (not secret; NULL → workspace EMAIL_TO)"
bool auto_resolve_incidents "auto-resolve on pass (default true, ADR 0034)"
}
notification_channels {
uuid id PK
string name "admin-chosen label"
string type "teams / slack / email"
string webhook_secret_ref "teams/slack - SecretStore key"
string email_recipients "email type - comma-separated addresses, not secret"
uuid created_by FK "SET NULL"
}
suite_notification_channels {
uuid suite_id PK,FK "CASCADE"
uuid channel_id PK,FK "RESTRICT - unlink before deleting a live channel"
}
workspace_health {
string key PK "signal name, e.g. orchestration_poll_staleness"
timestamptz alerted_at "delivered-first flag - set only after a publish succeeded"
}
privacy_settings {
int id PK "always 1 - singleton zero-sample privacy switch"
bool zero_sample_mode "default false - effective state is the env floor OR this"
uuid updated_by FK "the admin who last changed it; SET NULL on erasure"
}
scoring_settings {
int id PK "always 1 - singleton health-score weights (ADR 0005 amendment)"
numeric warn_weight "0 <= warn <= fail <= critical, critical > 0 (table CHECK)"
numeric fail_weight
numeric critical_weight "also the normaliser - all-critical scores 0"
uuid updated_by FK "the admin who last changed it; SET NULL on erasure"
}
coverage_settings {
int id PK "always 1 - singleton sensitivity for automatic baselines (ADR 0047)"
jsonb config "per-baseline thresholds + baseline shape; absent row = built-in defaults"
uuid updated_by FK "the admin who last changed it; SET NULL on erasure"
}
telemetry_settings {
int id PK "always 1 - singleton usage-telemetry opt-in (ADR 0050)"
bool enabled "default false - absent row = off"
uuid updated_by FK "the admin who last changed it; SET NULL on erasure"
}
seasonal_periods {
uuid id PK
string name "unique case-insensitively (lower(name) index)"
string kind "recurring (month-and-day range every year, may wrap the year end) / dated"
smallint start_month "recurring only, with start_day / end_month / end_day"
date start_date "dated only, with end_date >= start_date"
string series "dated only - entries sharing one are occurrences of one period (ADR 0048)"
uuid created_by FK "SET NULL on erasure"
uuid updated_by FK "SET NULL on erasure"
}
llm_settings {
int id PK "always 1 - singleton provider config (ADR 0042)"
string provider "anthropic / openai_compatible"
string base_url "required for openai_compatible; the credential's destination"
string model
string api_key_secret_ref "SecretStore ref - the key itself is never stored"
string structured_output "native / prompt_json"
bool enabled "default false - no row or disabled means no LLM features"
}
llm_invocations {
uuid id PK
string kind "ping / sql_generation / check_suggestion / rca_narrative"
string status "pending / running / succeeded / failed"
uuid requested_by_user_id FK "SET NULL - record outlives requester"
uuid suite_id FK "SET NULL - the cost record outlives its scope"
jsonb request "caller ask - never warehouse values"
string context_fingerprint "sha256 of the assembled prompt"
jsonb response
string error
int input_tokens "+ output_tokens, duration_ms - the cost record"
}
notifications {
uuid id PK
uuid user_id FK "CASCADE — the recipient's own feed state dies with the recipient"
uuid suite_id FK "CASCADE — feed entries die with the suite, like its runs"
uuid run_id FK "SET NULL — the entry outlives the run; the API then withholds the link"
uuid incident_id FK "SET NULL"
string severity "pass / warn / fail / critical / error (the run's worst tier)"
string title
text body "built from the redacted RunReport — check names only, never an observed value"
string deep_link "/results/<run_id>"
timestamptz read_at "NULL = unread; per user"
}
incidents {
uuid id PK
uuid asset_id FK "CASCADE (ADR 0034)"
uuid check_id FK "CASCADE — (asset,check) anchor; ≤1 active per pair (partial unique index)"
uuid suite_id FK "CASCADE — authz + routing"
string status "open / acknowledged / resolved"
string resolved_by "user / auto (NULL until resolved)"
string resolution "fixed / expected_change / false_positive (ADR 0047) — nullable"
int occurrence_count "repeat failures attach, not duplicate"
timestamptz last_seen_at
timestamptz acknowledged_at "+ acknowledge_note — lifecycle transition record"
uuid acknowledged_by FK "SET NULL"
timestamptz resolved_at "+ resolution_note"
uuid resolved_by_user_id FK "SET NULL"
uuid prior_incident_id FK "SET NULL — reopen chain"
jsonb evidence "deterministic layer-1 card (no sample rows — PII)"
}
users ||--o{ api_keys : "PATs (CASCADE — keys die with the user)"
users |o--o{ workspace_members : "invited_by (SET NULL)"
users ||--o{ sessions : "OTP browser sessions (CASCADE)"
users ||--o{ connections : "created_by"
users ||--o{ suites : "created_by"
users ||--o{ schedules : "created_by"
users ||--o{ shares : "grantee (CASCADE)"
users |o--o{ connection_versions : "changed_by (SET NULL)"
users |o--o{ check_versions : "changed_by (SET NULL)"
users |o--o{ audit_events : "actor_user_id (SET NULL)"
users |o--o{ llm_invocations : "requested_by (SET NULL)"
suites |o--o{ llm_invocations : "context scope (SET NULL)"
connections ||--o{ connection_versions : "config history (CASCADE)"
connections ||--o{ suites : "datasource for"
connections ||--o{ pipeline_runs : "orchestrator connection"
connections |o--o{ assets : "provenance hint (SET NULL)"
users |o--o{ assets : "owner_user_id (SET NULL)"
assets |o--o{ suites : "resolved target (SET NULL)"
assets |o--o{ runs : "stamped at dispatch (SET NULL)"
assets ||--o{ asset_column_classifications : "detected on (CASCADE, ADR 0049)"
users |o--o{ asset_column_classifications : "dismissed_by (SET NULL)"
assets ||--o{ lineage_edges : "upstream (CASCADE)"
assets ||--o{ lineage_edges : "downstream (CASCADE)"
connections ||--o{ lineage_edges : "refreshed by (CASCADE)"
suites ||--o{ checks : "contains (CASCADE)"
contracts |o--o{ suites : "enforced by (SET NULL)"
connections |o--o{ checks : "comparison source (ADR 0015, RESTRICT)"
suites ||--o{ runs : "executed as (CASCADE)"
suites ||--o{ shares : "shared via (CASCADE)"
suites ||--o{ trigger_bindings : "triggered by (CASCADE)"
suites ||--o{ schedules : "scheduled by (CASCADE)"
suites ||--o| suite_notifications : "alert config (CASCADE)"
suites ||--o{ suite_notification_channels : "linked channels (CASCADE)"
notification_channels ||--o{ suite_notification_channels : "referenced by (RESTRICT)"
users |o--o{ notification_channels : "created_by (SET NULL)"
checks ||--o| monitor_baselines : "diff reference (CASCADE)"
users |o--o{ monitor_baselines : "captured_by (SET NULL)"
suites ||--o{ check_suggestions : "review queue (CASCADE)"
checks |o--o{ check_suggestions : "created on accept (SET NULL)"
users |o--o{ check_suggestions : "decided_by (SET NULL)"
checks ||--o{ check_versions : "config history (CASCADE)"
suites ||--o{ suite_tags : "labelled (CASCADE)"
assets ||--o{ asset_tags : "labelled (CASCADE)"
connections ||--o{ connection_tags : "labelled (CASCADE)"
checks ||--o{ check_tags : "labelled (CASCADE)"
tags ||--o{ suite_tags : "assigned to (CASCADE)"
tags ||--o{ asset_tags : "assigned to (CASCADE)"
tags ||--o{ connection_tags : "assigned to (CASCADE)"
tags ||--o{ check_tags : "assigned to (CASCADE)"
checks ||--o{ results : "evaluated as (CASCADE)"
runs ||--o{ results : "produces (CASCADE)"
assets ||--o{ incidents : "anchored to (CASCADE, ADR 0034)"
checks ||--o{ incidents : "anchored to (CASCADE)"
suites ||--o{ incidents : "authz + routing (CASCADE)"
incidents |o--o| incidents : "reopen chain — prior_incident_id (SET NULL)"
users ||--o{ notifications : "recipient (CASCADE)"
suites ||--o{ notifications : "scope (CASCADE)"
runs |o--o{ notifications : "deep-link anchor (SET NULL)"
incidents |o--o{ notifications : "deep-link anchor (SET NULL)"
pipeline_runs ||..o{ runs : "triggered_by marker (no FK)"
trigger_bindings }o..o{ pipeline_runs : "(provider, pipeline, env) match (no FK)"Reading notes¶
- Conventions (elided from the diagram for noise): every table has a
gen_random_uuid()UUID PK andcreated_at; user-editable entities also carryupdated_at(runs/resultsdeliberately don't — they are engine-mutated event rows whose lifecycle lives instarted_at/finished_at/sample_failures_purged_at). Status/type columns areTEXT+CHECKconstraints, not native PG enums (migration ergonomics). shares.permissiongrants areview/editonly.adminis legal in the DBCHECKbut never granted (share_service.GRANTABLE_PERMISSIONS): workspace-admin is implicit on every suite (a storedusers.role = 'admin', or theWORKSPACE_ADMIN_EMAILSbootstrap/break-glass allowlist — ADR 0033, amending ADR 0027) andownerissuites.created_by, not a share. Authorization is two axes: the stored workspace role gates whole resource classes (connection mutations are Admin-only), the per-suite grant gates individual suites; neither replaces the other.- Cascade posture (ADR 0020): deleting a suite cascades its checks, runs, results, shares, trigger bindings, schedules, and notification config; deleting a connection cascades its version history. History is not retained past entity deletion — accepted. Version snapshots survive their author (
changed_byisSET NULL), not their entity. pipeline_runs≠runs— no FK between them. Orchestrator pipeline executions correlate to the DQ suite runs they trigger only via the string markerruns.triggered_by = '<provider>:<pipeline_or_dag_id>:<provider_run_id>'(dotted lines above);trigger_bindingsmatches pipeline runs by(provider, pipeline_or_dag_id, env), also without an FK. A partial unique index onruns (suite_id, triggered_by)dedupes orchestration-triggered runs.- Singleton constraints: at most one orchestrator connection per
(type, env)(partial unique index overadf/airflow/dbtonly — datasources may repeat); onesuite_notificationsrow per suite; one livesharesrow per(suite, user). - Secrets are never in these tables.
connections.secret_ref/suite_notifications.webhook_secret_refhold SecretStore keys; version snapshots deliberately omit credentials, so a credential rotation records no version.
Runtime flows¶
The diagrams above show structure (who talks to whom, what is stored); these two show ordering for the flows that cross the most components.
Suite run lifecycle¶
Every run — manual (POST /suites/{id}/run), scheduled (the 60s beat dispatcher), or orchestration-triggered — converges on the same path once the Run row exists:
sequenceDiagram
autonumber
participant Trig as Trigger<br/>(manual API · schedule beat · pipeline event)
participant API as FastAPI / beat dispatcher
participant PG as PostgreSQL
participant BR as Redis broker
participant W as Celery worker (run_suite)
participant KV as SecretStore (Key Vault)
participant DS as Datasource
participant Pub as ResultPublisher (Teams · Slack · email)
Trig->>API: trigger suite run
API->>PG: INSERT runs (status=queued, triggered_by marker)
API->>BR: send_task run_suite(run_id)
Note over API,BR: broker down → run marked terminal failed,<br/>never left stuck queued
API->>PG: store celery_task_id (enables cancel/revoke)
BR->>W: deliver task
W->>PG: load run — already cancelled? stop (cooperative cancel)
W->>PG: load suite + connection + checks, resolve target
W->>KV: get connection credential (secret_ref)
W->>W: build CheckRunner by connection.type (registry, ADR 0011)
Note over W,KV: any setup failure → run terminal failed
W->>DS: materialize flat-file batch path
Note over W,DS: batch absent → every check skip,<br/>run still succeeds
W->>PG: UPDATE runs SET status=running
W->>DS: execute by check.kind (ADR 0012) —<br/>expectation → GX validate · freshness/volume → monitor SQL
DS-->>W: one CheckOutcome per check
W->>W: severity.resolve_status — band unexpected-% against<br/>warn/fail/critical thresholds (ADR 0005/0016)
W->>PG: INSERT results (status, metric_value, redacted sample_failures)
W->>PG: UPDATE runs SET status=succeeded<br/>(failed = the adapter raised)
W->>Pub: publish_run_outcome — snooze suppression → dedup →<br/>publish (severity/alert_on routing inside the publisher)
Note over W,Pub: best-effort — a publish failure never<br/>affects the persisted runKey sources: worker/tasks.py (run_suite), services/run_service.py (execute_run), services/run_dispatch.py, services/severity.py, alerting/dispatch.py.
Orchestration event flow¶
How an ADF / Airflow / dbt pipeline outcome becomes (at most) a triggered suite run. Everything goes through the OrchestrationProvider seam (ADR 0004) — no provider branching:
sequenceDiagram
autonumber
participant ORC as ADF / Airflow / dbt
participant FE as Frontend nginx (public ingress)
participant API as FastAPI (internal)
participant KV as SecretStore (Key Vault)
participant PG as PostgreSQL
participant BR as Redis broker
participant W as Celery worker
Note over ORC: pipeline / DAG / dbt build reaches a terminal state
ORC->>FE: POST /api/v1/orchestration/events/{provider}<br/>(ADF — Azure Monitor alert · Airflow — on_*_callback · dbt — post-build callback)
FE->>API: same-origin proxy
API->>KV: load receiver secret
API->>API: authenticate — ADF constant-time token query param (ADR 0006)<br/>· Airflow HMAC-SHA256 over raw body (ADR 0007)
API->>API: OrchestrationProvider.parse → RunUpdate or AlertPing
alt AlertPing — run-anonymous Azure Monitor alert
API->>BR: enqueue targeted poll-now (provider + resource)
API-->>ORC: 200 (body status = reconciling)
else RunUpdate
API->>PG: upsert pipeline_runs ON (provider, provider_run_id)
alt status = succeeded
API->>PG: match enabled trigger_bindings (provider, pipeline_or_dag_id, env)
API->>PG: INSERT runs, triggered_by = provider:pipeline:run_id<br/>ON CONFLICT DO NOTHING (dedup index)
API->>BR: dispatch run_suite per triggered run
else status = failed
API->>API: alert the user only — failures never trigger suites
end
API-->>ORC: 200 (body status = recorded)
end
Note over W,PG: polling fallback — a 10-min beat sweeps every orchestrator connection<br/>via the provider REST API (list_recent_runs, 15-min lookback) into the SAME<br/>upsert + trigger path — a 30-min gap-recovery sweep (1-hour window) covers downtimeKey sources: api/v1/orchestration.py, services/orchestration_service.py, worker/tasks.py (poll_orchestration_runs, recover_orchestration_gaps).
Status semantics¶
Run lifecycle¶
runs.status describes execution, not data quality — a run whose checks all failed is still succeeded.
stateDiagram-v2
[*] --> queued : created (manual · schedule · trigger)
queued --> running : worker picks up
queued --> cancelled : user cancel
queued --> failed : dispatch/setup failure · reaper
running --> succeeded : execution completed
running --> cancelled : user cancel
running --> failed : adapter raised · reaper
succeeded --> [*]
failed --> [*]
cancelled --> [*]succeededmeans executed — checks may still have failed; the data-quality outcome lives inresults.status.- Cancel works on any non-terminal run: the API sets
cancelledand best-effort revokes the Celery task; the worker also honours the status cooperatively (start-check before executing). - The reaper drives runs orphaned in
queued/runningpast a threshold (task never published, or the worker died mid-run) to terminalfailed.
Result status derivation¶
results.status has two orthogonal families: the four severity tiers (ADR 0005) and the two operational statuses. Only the tiers carry health-score weight (0.5 / 1.0 / 2.0 for warn / fail / critical); skip/error must be excluded from the health-score denominator.
flowchart TD
O["CheckOutcome (from the runner)"] --> E{"runner could evaluate it?"}
E -- "no — evaluation raised" --> ERR["error — operational<br/>no tier · no metric · error message in observed_value"]
E -- yes --> S{"flat-file batch landed?<br/>(decided before execution)"}
S -- no --> SKIP["skip — operational<br/>not evaluated at all"]
S -- yes --> M{"thresholds set AND a<br/>bandable metric_value exists?"}
M -- "no — binary fallback (ADR 0005)" --> BIN{"GX success?"}
BIN -- yes --> PASS[pass]
BIN -- no --> FAIL[fail]
M -- yes --> BAND["band unexpected-% (0-100, higher = worse)<br/>against warn / fail / critical thresholds (ADR 0016)"]
BAND --> TIER["pass / warn / fail / critical<br/>thresholds are policy — they OVERRIDE GX success"]
classDef op fill:#F1EFE8,stroke:#5F5E5A,color:#444441
classDef tier fill:#E6F1FB,stroke:#185FA5,color:#0C449C
class ERR,SKIP op
class PASS,FAIL,TIER tierThe single decision lives in services/severity.py (resolve_status), shared by run persistence and the check-editor dry-run so a preview can never disagree with the run it previews.
Trust boundaries & authentication¶
The one-diagram consolidation of ADRs 0006 / 0007 / 0008 / 0028 — what crosses each boundary and what credential it carries:
flowchart LR
subgraph internet["🌍 Untrusted — public internet"]
B["Browser (React SPA)"]
AI["AI clients (MCP)"]
WH["ADF / Airflow / dbt webhooks"]
end
IDP["🔑 OIDC authority<br/>(Azure AD or AWS Cognito behind the generic DATAQ_AUTH_* contract)"]
subgraph aca["🟦 Trust boundary — Container Apps env"]
FE["Frontend nginx — SOLE public ingress (TLS)<br/>serves the SPA + runtime auth config (non-secret)"]
API["FastAPI — internal ingress only<br/>validates EVERY request itself (no EasyAuth)"]
WK["Celery worker — no ingress"]
BT["Celery beat — no ingress, own process"]
end
subgraph backing["🔒 Credentialed backing services"]
KV["Key Vault (all connection + webhook secrets)"]
PG[("PostgreSQL — dataq db,<br/>least-priv dataq_app role")]
RD[("Redis — password auth")]
APPI["App Insights (PII-redacted logs + traces)"]
end
subgraph egress["🌐 Outbound — credentials fetched from Key Vault per use"]
DS["Datasources — Snowflake · ADLS · S3 · GCS · Unity Catalog · Iceberg · PostgreSQL · MySQL · Trino · SQL Server · Athena · Redshift · BigQuery"]
AL["Teams / Slack webhooks · SMTP"]
OAPI["ADF / Airflow REST APIs (polling)"]
LLM["LLM provider — Anthropic / OpenAI-compat<br/>(off by default; masked profiler stats or a<br/>redacted observed_value only — never raw samples)"]
end
B -- "OIDC auth-code + PKCE" --> IDP
IDP -. "same bearer token, minted by the user's<br/>web-app sign-in and copied into the client<br/>(no client-driven flow — ADR 0008)" .-> AI
B -- "HTTPS · bearer JWT on /api" --> FE
AI -- "bearer JWT on /mcp" --> FE
WH -- "ADF: shared-secret token in URL (ADR 0006)<br/>Airflow / dbt: HMAC-SHA256 body signature (ADR 0007/0029)" --> FE
FE -- "same-origin proxy /api · /mcp · /healthz (HTTP/1.1)" --> API
API -- "JWT validated (fastapi-azure-auth / MCP JWTVerifier)<br/>+ per-suite authz + sample redaction" --> API
API -- "enqueue tasks" --> RD
RD -- "deliver tasks" --> WK
BT -- "schedule tick, enqueue tasks" --> RD
API -- "UAMI — no stored credential" --> KV
WK -- "UAMI — no stored credential" --> KV
API --> PG
WK --> PG
BT --> PG
API --> APPI
WK --> APPI
BT --> APPI
WK -- "GX checks / monitor SQL" --> DS
WK -- "run-outcome alerts" --> AL
WK -- "10-min poll fallback" --> OAPI
WK -- "sql_generation / check_suggestion / rca_narrative,<br/>via the dedicated llm Celery queue" --> LLM
classDef hub fill:#E6F1FB,stroke:#185FA5,color:#0C449C
classDef ext fill:#F1EFE8,stroke:#5F5E5A,color:#444441
classDef sec fill:#FAEEDA,stroke:#854F0B,color:#633806
classDef out fill:#E1F5EE,stroke:#0F6E56,color:#085041
class FE,API,WK,BT hub
class B,AI,WH,IDP ext
class KV,PG,RD,APPI sec
class DS,AL,OAPI,LLM out
style internet fill:#F7F6F2,stroke:#5F5E5A
style aca fill:#F4F9FE,stroke:#185FA5
style backing fill:#FDF7EC,stroke:#854F0B
style egress fill:#F0FBF7,stroke:#0F6E56Boundary notes:
- Defense in depth, not perimeter trust: the API validates every request's bearer JWT itself (
fastapi-azure-authfor REST,JWTVerifierfor MCP — same tenant/audience/scope) even though it is only reachable through the frontend proxy. Platform-level auth (SWA EasyAuth) is explicitly disabled. - The only endpoints that bypass user JWT auth are the three orchestration webhook receivers — ADF · Airflow · dbt, each with its own secret scheme, above — the email-OTP sign-in endpoints (
/auth/otp/request·/auth/otp/verify, rate-limited, ADR 0032),POST /auth/logout(session cookie only) and the health/readiness probes (/healthz·/readyz). Webhook secrets live in Key Vault and are compared constant-time; they are never logged. - Nothing secret is baked into images or served to the browser. The frontend's runtime
DATAQ_AUTH_*config is non-secret OIDC metadata (ADR 0028); all real secrets resolve at use-time from Key Vault via user-assigned managed identity. - MCP is fail-closed: without resolvable auth config the
/mcpmount does not come up at all (ADR 0008).
Key invariants¶
- The frontend (Container App on Azure, ECS Fargate task on AWS) is the sole public surface (ADR 0028 §5). It's one generic nginx image whose auth is injected at runtime (
DATAQ_AUTH_*→ generic OIDC, validated against Azure AD or AWS Cognito — no MSAL, nothing cloud-specific baked in), and it reverse-proxies/api+/mcp+/healthzsame-origin to the internal-ingress API. The API is not reachable directly from the internet; external orchestrator webhooks land on the frontend and are proxied through. - Orchestration providers (ADF · Airflow · dbt) are not datasources. They live in
pipeline_runs, notruns. Trigger bindings map(provider, pipeline_id, env) → suite_id. - Scheduled/triggered suite runs are Celery-only. FastAPI never enqueues GX itself for a full suite run; it dispatches a task. Exception — synchronous preview paths: the check dry-run (
POST /suites/{id}/checks/dryrun) and the column profiler (POST /suites/{id}/profile) run a single GX check / a profiling query against the datasource synchronously in a threadpool (persisting nothing) — interactive authoring aids, not scheduled runs. - The worker consumes two Celery queues.
run_suiteand the periodic beat-dispatched tasks stay on the defaultceleryqueue; the three LLM intelligence tasks (sql_generation·check_suggestion·rca_narrative) route to a dedicatedllmqueue, so a backlog of long-running suite runs never starves an LLM request behind it. Both queues are consumed by the same worker process (-Q celery,llm) — this is queue separation for fair scheduling, not a second worker deployment. - celery-beat is its own process/service, never embedded in the worker. Beat only schedules — it enqueues tasks onto the broker, it never executes one — so a worker OOM (overlapping large suites under
worker_concurrency=4and a 2 GiB hard limit) can never take the scheduler down with it. This is what stops orchestration polling, scheduled-suite dispatch, and every daily sweep from going silently dark because the process that ran them died. Exactly one beat instance may ever run (min=max=1 replica / desired_count=1) — a second would double-fire every periodic task. - The outbound LLM intelligence layer is live, off by default. An admin must configure a provider (Anthropic or an OpenAI-compatible endpoint) and credential before any call leaves the deployment. SQL generation and check suggestions send masked aggregate profiler statistics only; RCA narratives additionally send the triggering check's own
observed_value, routed through the same column-policy/warehouse-tag redaction floor every other results surface applies, plus itsexpected_value(a check-authored threshold, never masked). Raw sample rows are never sent on any path. See Security & data handling. - All connection secrets via the deployment's secret store in production / staging — Key Vault on Azure, Secrets Manager on AWS, OpenBao (KV v2) self-hosted (ADR 0039). Local dev may resolve secrets via
KV_SECRET_*env vars through theEnvSecretStorebackend (seebackend/app/core/secrets.py). No credentials are ever hardcoded. - The
/mcpendpoint exposes the same service layer to AI clients. The 59 FastMCP tools (35 read-only, 18 that change state, 6 live-probe tools gated like writes) are thin wrappers reusing the same services + per-suite authz + sample redaction as the REST API — no logic duplication. It mounts under either sign-in mode (SSO, email OTP) and stays unmounted, fail-closed, only when none is configured. Under SSO it validates the same OIDC bearer (Azure AD or Cognito — aJWTVerifieron the same tenant/audience/scope) or a PAT; under email OTP a PAT is the only accepted credential — a raw JWT and a session cookie are both rejected there, since there is no IdP-issued bearer to validate and a session is a browser-only credential. See ADR 0008 / ADR 0032. - Interactive API docs are off in production.
/docs,/redoc, and/openapi.jsonare disabled whenENVIRONMENT=prod(the prod-docs gate); available in dev/staging.