Trigger Bolt schedules from Airflow

Use the Paradime Airflow provider to trigger Bolt schedules from your Apache Airflow DAGs, wait for runs to complete, and download dbt™ run artifacts, so Bolt runs fit into your broader orchestration. The provider is published on the Astronomer Registry: airflow-provider-paradime-dbt.

Install

pip install airflow-provider-paradime-dbt

Account API keys need provider version 1.4.0 or later.

Create a connection

Account API key (recommended)
Workspace API key (legacy)

An account API key (prdm_cmp_...) can access one or more workspaces. One Airflow connection is enough to trigger schedules in all of them.

In Paradime, create an account API key with the Bolt schedules admin capability and select the workspaces it can access. Copy the key, the API endpoint, and the workspace UID listed next to each workspace.
In Airflow, create a connection of type Paradime with these fields:

Field	Value
API Endpoint	The API endpoint shown with the key
Account API Key	Your `prdm_cmp_...` key
Default Workspace UID	Optional. The workspace to use when a task does not set `workspace_uid`

Leave API Key (legacy) and API Secret (legacy) blank.To create the connection from an environment variable instead:

export AIRFLOW_CONN_PARADIME_DEFAULT='{
    "conn_type": "paradime",
    "extra": {
        "api_endpoint": "https://api.paradime.io/api/v1/<company_token>/graphql",
        "api_token": "prdm_cmp_...",
        "workspace_uid": "<default_workspace_uid>"
    }
}'

Every request with an account API key needs a workspace UID. Set it on the task with workspace_uid=, or set a default on the connection. The task value wins. If neither is set, the task fails with an error that says so.

Workspace API keys still work with no changes. A workspace key and secret are bound to one workspace, so you need one connection per workspace.

Generate your API key, secret, and endpoint from Paradime Workspace settings.
In Airflow, create a connection of type Paradime and fill in API Endpoint, API Key (legacy), and API Secret (legacy). Leave Account API Key blank.

workspace_uid has no effect on a connection that uses a workspace API key.

Reference the connection in your DAG with conn_id.

ParadimeBoltDbtScheduleRunOperator identifies the schedule with either slug (stable) or schedule_name, and also accepts commands, branch, retry_from_failure, and workspace_uid. ParadimeBoltDbtScheduleRunSensor and ParadimeBoltDbtScheduleRunArtifactOperator also accept workspace_uid. It is a templated field.

Trigger schedules in several workspaces

This DAG uses one account API key connection to trigger schedules in two workspaces, wait for the runs, and download their manifest.json. Each task sets workspace_uid to the workspace that owns its schedule.

from airflow.decorators import dag  # type: ignore[import]

from paradime_dbt_provider.operators.paradime import ParadimeBoltDbtScheduleRunArtifactOperator, ParadimeBoltDbtScheduleRunOperator
from paradime_dbt_provider.sensors.paradime import ParadimeBoltDbtScheduleRunSensor

PARADIME_CONN_ID = "your_paradime_conn_id"  # Update this to your connection id

# Map each Bolt schedule to the workspace that owns it
SCHEDULES = {
    "marketing": {"workspace_uid": "your_marketing_workspace_uid", "slug": "your-marketing-schedule-slug"},
    "finance": {"workspace_uid": "your_finance_workspace_uid", "slug": "your-finance-schedule-slug"},
}


@dag(
    default_args={"conn_id": PARADIME_CONN_ID},
)
def run_schedules_in_two_workspaces():
    for name, schedule in SCHEDULES.items():
        workspace_uid = schedule["workspace_uid"]

        # Run the schedule in its workspace and return the run id as the xcom return value
        task_run_schedule = ParadimeBoltDbtScheduleRunOperator(task_id=f"run_{name}", slug=schedule["slug"], workspace_uid=workspace_uid)

        # Get the run id from the xcom return value
        run_id = f"{{{{ task_instance.xcom_pull(task_ids='run_{name}') }}}}"

        # Wait for the run to complete. Use the same workspace UID as the task that triggered it.
        task_wait_for_schedule = ParadimeBoltDbtScheduleRunSensor(task_id=f"wait_for_{name}", run_id=run_id, workspace_uid=workspace_uid)

        # Download the manifest.json file from the run and return the path as the xcom return value
        task_download_manifest = ParadimeBoltDbtScheduleRunArtifactOperator(
            task_id=f"download_{name}_manifest",
            run_id=run_id,
            artifact_path="target/manifest.json",
            workspace_uid=workspace_uid,
        )

        task_run_schedule >> task_wait_for_schedule >> task_download_manifest


run_schedules_in_two_workspaces()

The sensor and the artifact operator must use the same workspace_uid as the task that triggered the run. A run id only exists in its own workspace.

Trigger a Bolt schedule

This DAG triggers a run for a Bolt schedule, checks the run’s status, and extracts the dbt™ artifacts. It needs no workspace_uid: use it with a workspace API key, or with an account API key that has a default workspace UID on the connection.

from airflow.decorators import dag  # type: ignore[import]

from paradime_dbt_provider.operators.paradime import ParadimeBoltDbtScheduleRunArtifactOperator, ParadimeBoltDbtScheduleRunOperator
from paradime_dbt_provider.sensors.paradime import ParadimeBoltDbtScheduleRunSensor

PARADIME_CONN_ID = "your_paradime_conn_id"  # Update this to your connection id
BOLT_SCHEDULE_SLUG = "your_schedule_slug"  # Update this to your schedule slug


@dag(
    default_args={"conn_id": PARADIME_CONN_ID},
)
def run_schedule_and_download_manifest():
    # Run the schedule and return the run id as the xcom return value
    task_run_schedule = ParadimeBoltDbtScheduleRunOperator(task_id="run_schedule", slug=BOLT_SCHEDULE_SLUG)

    # Get the run id from the xcom return value
    run_id = "{{ task_instance.xcom_pull(task_ids='run_schedule') }}"

    # Wait for the schedule to complete before continuing
    task_wait_for_schedule = ParadimeBoltDbtScheduleRunSensor(task_id="wait_for_schedule", run_id=run_id)

    # Download the manifest.json file from the schedule run and return the path as the xcom return value
    task_download_manifest = ParadimeBoltDbtScheduleRunArtifactOperator(task_id="download_manifest", run_id=run_id, artifact_path="target/manifest.json")

    # Get the path to the manifest.json file from the xcom return value
    output_path = "{{ task_instance.xcom_pull(task_ids='download_manifest') }}"

    task_run_schedule >> task_wait_for_schedule >> task_download_manifest


run_schedule_and_download_manifest()

Trigger a Bolt schedule with custom commands

This DAG overrides the dbt™ commands the run executes at runtime, then checks the run’s status and extracts the artifacts.

from airflow.decorators import dag  # type: ignore[import]

from paradime_dbt_provider.operators.paradime import ParadimeBoltDbtScheduleRunArtifactOperator, ParadimeBoltDbtScheduleRunOperator
from paradime_dbt_provider.sensors.paradime import ParadimeBoltDbtScheduleRunSensor

PARADIME_CONN_ID = "your_paradime_conn_id"  # Update this to your connection id
BOLT_SCHEDULE_SLUG = "your_schedule_slug"  # Update this to your schedule slug


@dag(
    default_args={"conn_id": PARADIME_CONN_ID},
)
def run_schedule_with_custom_commands():
    # Define the custom commands to run
    custom_commands = ["dbt run", "dbt test"]

    # Run the schedule with custom commands and return the run id as the xcom return value
    task_run_schedule = ParadimeBoltDbtScheduleRunOperator(task_id="run_schedule", slug=BOLT_SCHEDULE_SLUG, commands=custom_commands)

    # Get the run id from the xcom return value
    run_id = "{{ task_instance.xcom_pull(task_ids='run_schedule') }}"

    # Wait for the schedule to complete before continuing
    task_wait_for_schedule = ParadimeBoltDbtScheduleRunSensor(task_id="wait_for_schedule", run_id=run_id)

    # Download the manifest.json file from the schedule run and return the path as the xcom return value
    task_download_manifest = ParadimeBoltDbtScheduleRunArtifactOperator(task_id="download_manifest", run_id=run_id, artifact_path="target/manifest.json")

    # Get the path to the manifest.json file from the xcom return value
    output_path = "{{ task_instance.xcom_pull(task_ids='download_manifest') }}"

    task_run_schedule >> task_wait_for_schedule >> task_download_manifest


run_schedule_with_custom_commands()

Migrate from the dbt Cloud™ operators

If your DAGs trigger dbt Cloud™ jobs with the apache-airflow-providers-dbt-cloud provider, the DAG structure stays the same: swap each operator for its Paradime equivalent, and the job ID for a schedule slug.

dbt Cloud™ provider	Paradime provider
`DbtCloudRunJobOperator(job_id=...)`	`ParadimeBoltDbtScheduleRunOperator(slug=..., workspace_uid=...)`
`DbtCloudJobRunSensor(run_id=...)`	`ParadimeBoltDbtScheduleRunSensor(run_id=..., workspace_uid=...)`
`DbtCloudGetJobRunArtifactOperator(run_id=..., path=...)`	`ParadimeBoltDbtScheduleRunArtifactOperator(run_id=..., artifact_path=...)`
`dbt_cloud_conn_id`	`conn_id`
`account_id`	Not needed. The connection’s API endpoint already names your account
`steps_override`	`commands`
`retry_from_failure=True`	`retry_from_failure=True`

A few things work differently:

The run operator never waits. ParadimeBoltDbtScheduleRunOperator triggers the run and returns its run ID as the XCom value, like DbtCloudRunJobOperator with wait_for_termination=False. To wait for the run, add ParadimeBoltDbtScheduleRunSensor.
The sensor fails on a failed run. It succeeds when the run succeeds, and fails the task when the run fails, errors, is canceled, or is skipped. It is a standard Airflow sensor, so timeout, poke_interval, mode, and soft_fail work as usual.
Trigger rules, retries, and dependencies stay in Airflow. trigger_rule, task retries and retry_delay, and the order of your tasks behave exactly as before.
A sensor timeout doesn’t stop the run. When the sensor times out, the Bolt run carries on. Bolt has no job-level timeout: set sla_minutes on the schedule to be alerted when a run takes too long.
Retry from the point of failure. With retry_from_failure=True, the operator resumes the schedule’s last failed run from the failed command instead of starting a new run.

Before and after

A trigger-and-sensor pair on dbt Cloud™:

from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.providers.dbt.cloud.sensors.dbt import DbtCloudJobRunSensor

trigger_job = DbtCloudRunJobOperator(
    task_id="trigger_dbt_job",
    dbt_cloud_conn_id="dbt_cloud",
    account_id=12345,
    job_id=67890,
    wait_for_termination=False,
)

wait_for_job = DbtCloudJobRunSensor(
    task_id="wait_for_dbt_job",
    dbt_cloud_conn_id="dbt_cloud",
    account_id=12345,
    run_id=trigger_job.output,
    timeout=4 * 60 * 60,
)

trigger_job >> wait_for_job

The same pair on Paradime:

from paradime_dbt_provider.operators.paradime import ParadimeBoltDbtScheduleRunOperator
from paradime_dbt_provider.sensors.paradime import ParadimeBoltDbtScheduleRunSensor

trigger_job = ParadimeBoltDbtScheduleRunOperator(
    task_id="trigger_dbt_job",
    conn_id="paradime",
    slug="your-schedule-slug",
    workspace_uid="your_workspace_uid",
)

wait_for_job = ParadimeBoltDbtScheduleRunSensor(
    task_id="wait_for_dbt_job",
    conn_id="paradime",
    run_id=trigger_job.output,
    workspace_uid="your_workspace_uid",
    timeout=4 * 60 * 60,
)

trigger_job >> wait_for_job

Switch one project at a time

To move projects over one at a time, keep both setups in the DAG and choose one per project with a flag. Switching a project, or rolling it back, is then a one-line change. The task IDs are the same on both sides, so the DAG’s graph and history don’t change.

from airflow.decorators import dag  # type: ignore[import]
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.providers.dbt.cloud.sensors.dbt import DbtCloudJobRunSensor

from paradime_dbt_provider.operators.paradime import ParadimeBoltDbtScheduleRunOperator
from paradime_dbt_provider.sensors.paradime import ParadimeBoltDbtScheduleRunSensor

DBT_CLOUD_CONN_ID = "dbt_cloud"
PARADIME_CONN_ID = "paradime"

# Which product runs each project, and where its job lives in each
PROJECTS = {
    "staging": {
        "platform": "paradime",  # "dbt_cloud" or "paradime"
        "dbt_cloud_job_id": 67890,
        "workspace_uid": "your_staging_workspace_uid",
        "slug": "your-staging-schedule-slug",
    },
    "marts": {
        "platform": "dbt_cloud",
        "dbt_cloud_job_id": 67891,
        "workspace_uid": "your_marts_workspace_uid",
        "slug": "your-marts-schedule-slug",
    },
}


def run_dbt_job(project):
    config = PROJECTS[project]

    if config["platform"] == "paradime":
        trigger = ParadimeBoltDbtScheduleRunOperator(
            task_id=f"trigger_{project}",
            conn_id=PARADIME_CONN_ID,
            slug=config["slug"],
            workspace_uid=config["workspace_uid"],
            trigger_rule="all_done",
        )
        wait = ParadimeBoltDbtScheduleRunSensor(
            task_id=f"wait_for_{project}",
            conn_id=PARADIME_CONN_ID,
            run_id=trigger.output,
            workspace_uid=config["workspace_uid"],
            timeout=4 * 60 * 60,
            trigger_rule="all_done",
        )
    else:
        trigger = DbtCloudRunJobOperator(
            task_id=f"trigger_{project}",
            dbt_cloud_conn_id=DBT_CLOUD_CONN_ID,
            job_id=config["dbt_cloud_job_id"],
            wait_for_termination=False,
            trigger_rule="all_done",
        )
        wait = DbtCloudJobRunSensor(
            task_id=f"wait_for_{project}",
            dbt_cloud_conn_id=DBT_CLOUD_CONN_ID,
            run_id=trigger.output,
            timeout=4 * 60 * 60,
            trigger_rule="all_done",
        )

    trigger >> wait
    return trigger, wait


@dag(schedule="15 1 * * *", catchup=False)
def daily_dbt():
    previous_wait = None
    for project in PROJECTS:
        trigger, wait = run_dbt_job(project)
        if previous_wait is not None:
            previous_wait >> trigger
        previous_wait = wait


daily_dbt()

Cut over

When you migrate step by step, switch the project’s jobs to Bolt and change its flag in the same window:

Switch the project’s groups to Bolt on the migration page. This turns off the jobs’ schedules and Git triggers in dbt Cloud™.
Set the project’s platform to "paradime" and deploy the DAG.

The order matters because a dbt Cloud™ job with its triggers turned off still runs when the API calls it. Until the DAG changes, Airflow keeps starting the job in dbt Cloud™. To roll back, set platform back to "dbt_cloud", deploy the DAG, and roll the group back on the migration page.

Schedule Anatomy

Deploy Schedules

Managing Schedules

CI/CD

Trigger Bolt schedules from Airflow

Install

Create a connection

Trigger schedules in several workspaces

Trigger a Bolt schedule

Trigger a Bolt schedule with custom commands

Migrate from the dbt Cloud™ operators

Before and after

Switch one project at a time

Cut over

​Install

​Create a connection

​Trigger schedules in several workspaces

​Trigger a Bolt schedule

​Trigger a Bolt schedule with custom commands

​Migrate from the dbt Cloud™ operators

​Before and after

​Switch one project at a time

​Cut over

Install

Create a connection

Trigger schedules in several workspaces

Trigger a Bolt schedule

Trigger a Bolt schedule with custom commands

Migrate from the dbt Cloud™ operators

Before and after

Switch one project at a time

Cut over