Skip to main content

Run checks from any orchestrator

Goal​

You'll run Measured Cloud validations as a step of a pipeline, so a failed check stops the steps after it. This page covers a shell or cron job, Prefect and Snowflake tasks. For Airflow and Dagster, use their packages: Use with Airflow, Use with Dagster.

It also shows the other direction: Measured Cloud telling your pipeline that a run finished, so the next step starts when the checks pass.

Every recipe follows the same pattern: start a run, wait for it, and fail the step unless the run succeeded. A skipped run checked nothing (for example, your organization's monthly runs were used up), so it fails the step too.

Prereqs​

  • A service account token (dqsa_…) with the editor role on the workspace. A pipeline shouldn't depend on a person's API key.
  • The validation's id, and your organization and workspace ids. Every data asset with rules has a default validation that holds all of them.

A shell script or cron job​

The measuredcloud command line tool, from the Python SDK, does the whole pattern and exits non-zero when the checks fail:

export MEASUREDCLOUD_URL=https://app.measured.cloud
export MEASUREDCLOUD_API_KEY=dqsa_...
export MEASUREDCLOUD_ORG_ID=<org id>
export MEASUREDCLOUD_WORKSPACE_ID=<workspace id>

measuredcloud validations run <validation id> --wait --junit report.xml && ./load_downstream.sh

--junit writes a report with one test per rule, for CI systems that show test results.

Status: preview. The SDK isn't on PyPI yet (#512). Until it is, use the API directly:

base="https://app.measured.cloud/api/v2/organizations/$ORG_ID/workspaces/$WS_ID"
auth="Authorization: Bearer $MEASUREDCLOUD_API_KEY"
run=$(curl -sf -X POST -H "$auth" -H "Content-Type: application/json" \
  -H "Idempotency-Key: cron-$(date +%F)-orders" -d '{"data": {}}' \
  "$base/validations/$VALIDATION_ID/run" | jq -r .data.execution_run_id)
while :; do
  status=$(curl -s -H "$auth" "$base/execution-runs/$run" | jq -r '.data.status // "pending"')
  case "$status" in
    succeeded) break ;;
    failed|error|cancelled|skipped) echo "validation $VALIDATION_ID: $status" >&2; exit 1 ;;
  esac
  sleep 10
done

The Idempotency-Key makes a rerun of the same job on the same day return the run it already started instead of starting another; leave it out to always start a new run.

Prefect​

A task that raises fails, and Prefect doesn't run the tasks that depend on it:

import os
import time

import requests
from prefect import flow, task
from prefect.runtime import task_run

BASE = (
    "https://app.measured.cloud/api/v2/organizations/"
    f"{os.environ['MEASUREDCLOUD_ORG_ID']}/workspaces/{os.environ['MEASUREDCLOUD_WORKSPACE_ID']}"
)


@task(retries=2, retry_delay_seconds=30)
def check_data_quality(validation_id: str) -> dict:
    headers = {
        "Authorization": f"Bearer {os.environ['MEASUREDCLOUD_API_KEY']}",
        # The same task run keeps its id across retries, so a retry waits for
        # the run the first attempt started instead of starting another.
        "Idempotency-Key": f"prefect:{task_run.id}",
    }
    ack = requests.post(
        f"{BASE}/validations/{validation_id}/run", json={"data": {}}, headers=headers, timeout=30
    )
    ack.raise_for_status()
    run_id = ack.json()["data"]["execution_run_id"]
    while True:
        resp = requests.get(f"{BASE}/execution-runs/{run_id}", headers=headers, timeout=30)
        if resp.status_code != 404:  # 404: the run isn't recorded yet
            resp.raise_for_status()
            run = resp.json()["data"]
            if run["status"] == "succeeded":
                return run
            if run["status"] not in ("pending", "running"):
                raise RuntimeError(f"validation {validation_id}: {run['status']}")
        time.sleep(10)


@flow
def nightly(validation_id: str = "00000000-0000-0000-0000-0000000000aa"):
    loaded = load_orders()
    checked = check_data_quality(validation_id, wait_for=[loaded])
    publish_dashboards(wait_for=[checked])

Keep the token in a Prefect Secret block or your deployment's environment, not in the flow's code.

Snowflake tasks​

A Snowflake stored procedure can call Measured Cloud through an external access integration. Run it as a task after the task that loads the data: when the checks fail, the procedure raises, the task fails and the tasks after it don't run.

  1. Let Snowflake reach Measured Cloud, and store the token. This needs a role that can create integrations, such as ACCOUNTADMIN:

    CREATE OR REPLACE NETWORK RULE measuredcloud_api
      MODE = EGRESS TYPE = HOST_PORT VALUE_LIST = ('app.measured.cloud');

    CREATE OR REPLACE SECRET measuredcloud_token
      TYPE = GENERIC_STRING SECRET_STRING = '<dqsa_… token>';

    CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION measuredcloud_access
      ALLOWED_NETWORK_RULES = (measuredcloud_api)
      ALLOWED_AUTHENTICATION_SECRETS = (measuredcloud_token)
      ENABLED = TRUE;

  2. Create the procedure:

    CREATE OR REPLACE PROCEDURE run_measuredcloud_validation(validation_id STRING)
      RETURNS STRING
      LANGUAGE PYTHON
      RUNTIME_VERSION = '3.11'
      PACKAGES = ('snowflake-snowpark-python', 'requests')
      HANDLER = 'run'
      EXTERNAL_ACCESS_INTEGRATIONS = (measuredcloud_access)
      SECRETS = ('token' = measuredcloud_token)
    AS
    $$
    import time

    import _snowflake
    import requests

    BASE = "https://app.measured.cloud/api/v2/organizations/<org id>/workspaces/<workspace id>"


    def run(session, validation_id):
        headers = {"Authorization": "Bearer " + _snowflake.get_generic_secret_string("token")}
        ack = requests.post(
            f"{BASE}/validations/{validation_id}/run", json={"data": {}}, headers=headers, timeout=30
        )
        ack.raise_for_status()
        run_id = ack.json()["data"]["execution_run_id"]
        while True:
            resp = requests.get(f"{BASE}/execution-runs/{run_id}", headers=headers, timeout=30)
            if resp.status_code != 404:
                resp.raise_for_status()
                status = resp.json()["data"]["status"]
                if status == "succeeded":
                    return f"validation {validation_id}: succeeded"
                if status not in ("pending", "running"):
                    raise Exception(f"validation {validation_id}: {status}")
            time.sleep(10)
    $$;

  3. Run it after the load:

    CREATE OR REPLACE TASK check_orders
      WAREHOUSE = transform_wh
      AFTER load_orders
    AS
      CALL run_measuredcloud_validation('<validation id>');

    Tasks that run AFTER check_orders only run when the checks passed.

The procedure waits while Measured Cloud runs the checks, so the task's warehouse stays up for that time. For a long validation, start the run from the task and let the run-finished event trigger what comes next instead.

Start the next step when a run finishes​

Measured Cloud can tell your pipeline when a run finishes, whoever started it: a schedule, a person, an API call. Set up the outbound webhook in Settings → Catalog sync (see Integrate with your data catalog). Despite the name, it carries every event, including one execution_run.completed per finished run:

{
  "event": "execution_run.completed",
  "execution_run": {
    "id": "…",
    "status": "succeeded",
    "validation": {"id": "…", "name": "orders nightly"},
    "health_score": 98.0
  }
}

The full event is in A run finished. A receiver checks the signature, then acts on the event:

import hashlib
import hmac
import json
import os

SECRET = os.environ["MEASUREDCLOUD_WEBHOOK_SECRET"].encode()


def handle(body: bytes, signature_header: str) -> None:
    expected = "sha256=" + hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature_header or ""):
        raise PermissionError("bad signature")
    event = json.loads(body)
    if event["event"] != "execution_run.completed":
        return  # result and certification events
    run = event["execution_run"]
    if run["validation"] and run["validation"]["name"] == "orders nightly":
        if run["status"] == "succeeded":
            trigger_downstream()  # e.g. start a dbt job, refresh a dashboard
        else:
            page_the_data_team(run)