Bolt - Paradime

Prerequisites:

This feature is available with the Paradime Bolt plan.
Your API keys must have either Bolt Schedules Admin or Bolt Schedules Metadata Viewer capabilities.

These examples authenticate with an account API key (api_secret="prdm_cmp_..." plus workspace_uid), which requires paradime-io 6.0.0 or later. Legacy workspace API keys (api_key + api_secret) are still supported. See Getting Started.

The Bolt module allows you to easily manage and control Bolt schedules and runs within your workspace. It provides tools to create, configure, and monitor schedules, automate tasks, and access detailed logs and reports.

Schedules are identified by slug. Every Bolt method that targets a schedule accepts a slug= keyword argument — the identifier returned by create_schedule (and shown in the Bolt UI).

List Bolt schedules

Get a list of Bolt schedules. The list is paginated. The total count of schedules is also returned.

Args
Returns

offset (int): The offset value for pagination. Default is 0.limit (int): The limit value for pagination. Default is 100.show_inactive (bool): Flag to indicate whether to return inactive schedules instead of active schedules. Default is False.

BoltSchedules: An object containing the list of Bolt schedules and the total count of schedules.

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# List all schedules
schedules = paradime.bolt.list_schedules().schedules

Get latest runs for a schedule

Args
Returns

slug (str): The schedule slug returned by create_schedule. Preferred over schedule_name.schedule_name (str): Deprecated alias for slug. Still accepted for backwards compatibility — emits a DeprecationWarning. Exactly one of slug or schedule_name must be provided.offset (int): The offset value for pagination. Default is 0. Must be >= 0.limit (int): The limit value for pagination. Default is 50. Must be between 1 and 1000.

BoltScheduleRuns: An object containing the list of Bolt runs with details like ID, status, actor, timestamps, and git information.

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Get the latest 10 runs for a schedule
recent_runs = paradime.bolt.list_runs(
    slug="flowing-pachy-wyy4fs",
    offset=0,
    limit=10
)

# Access the runs and metadata
for run in recent_runs.runs:
    print(f"Run ID: {run.id}")
    print(f"Status: {run.state}")
    print(f"Started: {run.start_dttm}")
    print(f"Actor: {run.actor}")
    if run.git_info.branch:
        print(f"Branch: {run.git_info.branch}")
    if run.git_info.pull_request_id:
        print(f"PR: {run.git_info.pull_request_id}")
    print("---")

# Get runs with pagination (next page)
next_runs = paradime.bolt.list_runs(
    slug="flowing-pachy-wyy4fs",
    offset=10,
    limit=10
)

# Get all recent runs (up to 100)
all_recent_runs = paradime.bolt.list_runs(
    slug="flowing-pachy-wyy4fs",
    limit=100
)

Create a Bolt schedule

Create a new Bolt schedule and get back its slug. The slug is the identifier you pass as slug= to every other Bolt method (trigger_run, get_schedule, delete_schedule, etc.).

Args
Returns

Requiredname (str): Human-readable schedule name shown in the Bolt UI.schedule (str): Cron expression (e.g. "0 1 * * *") or the literal "OFF" for manual-only runs.environment (str): Name of the environment to run in (e.g. "production").commands (List[str]): Commands the schedule should run, in order (e.g. ["dbt run", "dbt test"]).Optional scalarsgit_branch (Optional[str]): Git branch the run should check out. Defaults to the environment’s branch.description (Optional[str]): Free-text description shown in the UI.timezone (Optional[str]): IANA timezone for the cron expression (e.g. "UTC", "Europe/London").owner_email (Optional[str]): Email of the workspace member who should own the schedule.suspended (Optional[bool]): Create the schedule already suspended. Defaults to active.sla_seconds (Optional[int]): Soft SLA window in seconds; runs exceeding this are surfaced as overdue.trigger_on_merge (Optional[bool]): If True, run on every merge to git_branch.Optional nested objects (each accepts either a typed Pydantic input model or an equivalent dict)notifications (Optional[BoltNotificationsInput | dict]): Slack / Teams / email notification routing.integrations (Optional[BoltIntegrationsInput | dict]): PagerDuty / Datadog / incident.io / New Relic incident triggers fired on failures.self_healing (Optional[BoltSelfHealingConfigInput | dict]): Paradime self-healing agent (auto-retry + Slack updates).turbo_ci (Optional[BoltDeferredScheduleConfigInput | dict]): Turbo CI config — defer state from another schedule’s last successful run.deferred_schedule (Optional[BoltDeferredScheduleConfigInput | dict]): Slim-CI-style deferred schedule config.schedule_trigger (Optional[BoltScheduleTriggerInput | dict]): Run this schedule when a parent schedule (possibly in another workspace) finishes.env_vars (Optional[List[BoltEnvironmentVariableInput | dict]]): Environment-variable overrides for this schedule.

str: The slug assigned by the backend. Pass this value as slug= to every other Bolt method.

There is a short consistency window (~10s) between schedule creation and the trigger path accepting the new slug. Callers that immediately invoke trigger_run on a brand-new slug may need to retry for a few seconds.

Create a minimal schedule

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Create a manual-only schedule and capture its slug
slug = paradime.bolt.create_schedule(
    name="Nightly build",
    schedule="OFF",
    environment="production",
    commands=["dbt build"],
)
print(f"created schedule slug: {slug}")

Create a schedule with notifications, env vars, and self-healing

# First party modules
from paradime import Paradime
from paradime.apis.bolt.types import (
    BoltNotificationsInput,
    BoltNotificationChannelInput,
    BoltSelfHealingConfigInput,
    BoltEnvironmentVariableInput,
)

paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

slug = paradime.bolt.create_schedule(
    name="Nightly build",
    schedule="0 1 * * *",
    environment="production",
    commands=["dbt build"],
    git_branch="main",
    description="Nightly run of the production warehouse build.",
    timezone="UTC",
    notifications=BoltNotificationsInput(
        slack_notifications=[
            BoltNotificationChannelInput(channel="#data-alerts", events=["failed"]),
        ],
        email_notifications=[
            BoltNotificationChannelInput(channel="ops@example.com", events=["failed", "passed"]),
        ],
    ),
    self_healing=BoltSelfHealingConfigInput(enabled=True, slack_channel="#data-alerts"),
    env_vars=[
        BoltEnvironmentVariableInput(key="DBT_PROFILES_DIR", value="/workspace/profiles"),
        BoltEnvironmentVariableInput(key="DBT_TARGET", value="prod"),
    ],
)

Create a schedule using raw dicts (escape hatch)

Every nested input also accepts a plain dict that matches the GraphQL shape. Useful when you want to set a field that isn’t yet modelled as a typed input.

# First party modules
from paradime import Paradime

paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

slug = paradime.bolt.create_schedule(
    name="Nightly build",
    schedule="OFF",
    environment="production",
    commands=["dbt build"],
    notifications={
        "slackNotifications": [
            {"channel": "#data-alerts", "events": ["failed"]},
        ],
    },
    env_vars=[{"key": "DBT_TARGET", "value": "prod"}],
)

Delete a Bolt schedule

Delete a Bolt schedule by slug. Schedules defined in YAML cannot be deleted via the API — remove them from the repository instead.

Args
Returns

slug (str): The schedule slug returned by create_schedule (also shown in the Bolt UI).

None

# First party modules
from paradime import Paradime

paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the slug of the schedule to delete
BOLT_SCHEDULE_SLUG = "nightly-build-a1b2c3"

paradime.bolt.delete_schedule(BOLT_SCHEDULE_SLUG)

Triggering a Bolt run

Triggers a run for a given schedule slug.

Args
Returns

slug (str): The schedule slug to trigger the run for. Preferred over schedule_name.schedule_name (str): Deprecated alias for slug. Still accepted for backwards compatibility — emits a DeprecationWarning. Exactly one of slug or schedule_name must be provided.commands (Optional[List[str]]): The list of commands to execute in the run. This will override the commands defined in the schedule. Defaults to None.branch (Optional[str]): The branch or commit hash to run the commands on. Defaults to None.pr_number (Optional[int]): The pull request number to associate with the run. Defaults to None.reason (Optional[str]): A freeform reason/label describing why or from where the run was triggered (e.g. the application that made the call). Stored with the run for context and auditing. Defaults to None.

int: The ID of the triggered run.

Trigger a run with default commands and branch

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the Bolt schedule slug to trigger
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# Trigger a run of the Bolt schedule and get the run ID
run_id = paradime.bolt.trigger_run(slug=BOLT_SCHEDULE_SLUG)

# Get the run status
run_status = paradime.bolt.get_run_status(run_id)

Trigger a run with a custom git branch and PR Number

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the Bolt schedule slug to trigger
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# Replace with the branch name or commit hash
GIT_BRANCH = "feature-branch-123"

# Replace with the PR number to associate with the run (optional)
PR_NUMBER = 123

# Trigger a run of the Bolt schedule and get the run ID
run_id = paradime.bolt.trigger_run(slug=BOLT_SCHEDULE_SLUG, branch=GIT_BRANCH, pr_number=PR_NUMBER)

# Get the run status
run_status = paradime.bolt.get_run_status(run_id)

Trigger a run with custom commands

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the Bolt schedule slug to trigger
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# Replace with the commands to execute in the run
COMMANDS_OVERRIDE = ["dbt run --select order_items", "dbt test"]

# Trigger a run of the Bolt schedule and get the run ID
run_id = paradime.bolt.trigger_run(slug=BOLT_SCHEDULE_SLUG, commands=COMMANDS_OVERRIDE)

# Get the run status
run_status = paradime.bolt.get_run_status(run_id)

Trigger a run with a custom git branch

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the Bolt schedule slug to trigger
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# Replace with the branch name or commit hash
GIT_BRANCH = "feature-branch-123"

# Trigger a run of the Bolt schedule and get the run ID
run_id = paradime.bolt.trigger_run(slug=BOLT_SCHEDULE_SLUG, branch=GIT_BRANCH)

# Get the run status
run_status = paradime.bolt.get_run_status(run_id)

Trigger a run with a reason

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the Bolt schedule slug to trigger
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# A freeform label describing why or from where the run was triggered
REASON = "triggered by data-quality-bot"

# Trigger a run of the Bolt schedule and get the run ID
run_id = paradime.bolt.trigger_run(slug=BOLT_SCHEDULE_SLUG, reason=REASON)

# Get the run status
run_status = paradime.bolt.get_run_status(run_id)

Retrying a Bolt run

Retry the latest failed run of a schedule by slug Retries the latest failed run of a Bolt schedule by slug, without needing to know its run ID. Resumes from the failed command of the most recent run of the given schedule. The first failed dbt command is substituted with dbt retry when supported. Infrastructure commands (git clone, dbt deps) are skipped. A new Bolt run is created; the original run is unchanged.

Args
Returns

slug (str): The schedule slug whose latest failed run to retry. Preferred over schedule_name.schedule_name (str): Deprecated alias for slug. Still accepted for backwards compatibility — emits a DeprecationWarning. Exactly one of slug or schedule_name must be provided.

int: The ID of the newly created retry run.

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the slug of the schedule to retry
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# Retry the latest failed run; returns the new run ID
new_run_id = paradime.bolt.retry_schedule_from_failure(slug=BOLT_SCHEDULE_SLUG)

# Poll the new run like any other Bolt run
new_run_status = paradime.bolt.get_run_status(new_run_id)

Retry only failed commands by Run ID

Retries a failed Bolt run by re-running only the failed commands. The first failed dbt command is substituted with dbt retry when supported, so only the failed models are re-executed. Infrastructure commands (git clone, dbt deps) are skipped. A new Bolt run is created; the original run is unchanged.

Args
Returns

run_id (int): The ID of the failed run to retry.

int: The ID of the newly created retry run.

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the run ID of the failed Bolt run to retry
FAILED_BOLT_RUN_ID = 1

# Retry only the failed commands; returns the new run ID
new_run_id = paradime.bolt.retry_run(run_id=FAILED_BOLT_RUN_ID)

# Poll the new run like any other Bolt run
new_run_status = paradime.bolt.get_run_status(new_run_id)

Retry all commands by Run ID

Retries a Bolt run by re-running every original command verbatim, regardless of which ones succeeded or failed. Infrastructure commands (git clone, dbt deps) are excluded. A new Bolt run is created; the original run is unchanged.

Args
Returns

run_id (int): The ID of the run to retry.

int: The ID of the newly created retry run.

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the run ID of the Bolt run to retry
BOLT_RUN_ID = 1

# Re-run all original commands; returns the new run ID
new_run_id = paradime.bolt.retry_run_all(run_id=BOLT_RUN_ID)

# Poll the new run like any other Bolt run
new_run_status = paradime.bolt.get_run_status(new_run_id)

Streaming logs from a Bolt command

Tail the stdout and stderr of a Bolt command while it is still running, instead of waiting for the run to finish and reading the final logs. Two methods are available:

stream_command_logs(command_id) — a generator that yields lines as they arrive and stops automatically when the command finishes. Use this for live tailing.
get_command_logs(command_id, cursor) — a single batch fetch using an opaque cursor. Use this when you need finer control over polling cadence or want to interleave log polling with other work.

Both methods return BoltLogLine objects with a stream field (BoltLogStream.STDOUT or BoltLogStream.STDERR) and a line field (the raw log line).

To get a command_id, list the commands for a run with paradime.bolt.list_run_commands(run_id).

Stream logs until the command finishes

stream_command_logs polls until the command exits and yields each line as it arrives.

Args
Yields

command_id (int): The ID of the Bolt command to stream logs from.poll_interval (float): Seconds to wait between empty polls. Default is 2.0.

Iterator[BoltLogLine]: Each log line in arrival order within a poll batch. Stdout lines for the batch precede stderr lines (approximate interleaving — true cross-stream ordering is not recorded).

# Standard library modules
import os
import time

# First party modules
from paradime import Paradime
from paradime.apis.bolt.types import BoltLogStream, BoltRunState

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Trigger a run
run_id = paradime.bolt.trigger_run(slug="flowing-pachy-wyy4fs")

# Tail each command as it appears, until the run finishes
seen_command_ids: set[int] = set()
while True:
    for cmd in paradime.bolt.list_run_commands(run_id):
        if cmd.id in seen_command_ids:
            continue
        seen_command_ids.add(cmd.id)
        print(f"\n--- Command {cmd.id}: {cmd.command} ---")
        for log_line in paradime.bolt.stream_command_logs(cmd.id):
            prefix = "stderr" if log_line.stream is BoltLogStream.STDERR else "stdout"
            print(f"[{prefix}] {log_line.line}", end="")

    if paradime.bolt.get_run_status(run_id) is not BoltRunState.RUNNING:
        break
    time.sleep(2)

One-shot polling with a cursor

get_command_logs returns a single batch with an opaque cursor and a finished flag. Pass the cursor back into the next call to fetch only new lines.

Args
Returns

command_id (int): The ID of the Bolt command.cursor (str): Opaque cursor returned by the previous call. Use the default "0:0" on the first call to fetch from the beginning.

BoltCommandLogs: An object with lines (List[BoltLogLine]), cursor (str — pass to the next call), and finished (bool — flips to True once the command exits).

# First party modules
import time
from paradime import Paradime

paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

COMMAND_ID = 59241
cursor = "0:0"

while True:
    batch = paradime.bolt.get_command_logs(COMMAND_ID, cursor=cursor)

    for log_line in batch.lines:
        print(log_line.line, end="")

    if batch.finished:
        break

    cursor = batch.cursor
    if not batch.lines:
        time.sleep(2)

Cancelling a Bolt run

Cancels a Bolt run.

Args
Returns

run_id (int): The ID of the run to cancel.

none

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the run ID of the Bolt schedule to cancel
BOLT_SCHEDULE_RUN_ID = 1

# Cancel the run
paradime.bolt.cancel_run(run_id=BOLT_SCHEDULE_RUN_ID)

Getting a Bolt run artifacts

Get latest run manifest.json Retrieves the latest manifest JSON for a given schedule.

Args
Returns

slug (str): The schedule slug. Preferred over schedule_name.schedule_name (str): Deprecated alias for slug. Still accepted for backwards compatibility — emits a DeprecationWarning. Exactly one of slug or schedule_name must be provided.command_index (Optional[int]): The index of the command in the schedule. Defaults to None.

dict: The content of the latest manifest JSON.

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the Bolt schedule slug you want to get artifacts from
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# Get manifest.json dictionary
manifest_json = paradime.bolt.get_latest_manifest_json(slug=BOLT_SCHEDULE_SLUG)

Get latest run artifacts URL Retrieves the URL of the latest artifact for a given schedule.

Args
Returns

slug (str): The schedule slug. Preferred over schedule_name.schedule_name (str): Deprecated alias for slug. Still accepted for backwards compatibility — emits a DeprecationWarning. Exactly one of slug or schedule_name must be provided.artifact_path (str): The path of the artifact.command_index (Optional[int]): The index of the command in the schedule. Defaults to searching through all commands from the last command to the first.

str: The URL of the latest artifact.

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(api_endpoint="API_ENDPOINT", api_secret="prdm_cmp_...", workspace_uid="WORKSPACE_UID")

# Replace with the Bolt schedule slug you want to get artifacts from
BOLT_SCHEDULE_SLUG = "flowing-pachy-wyy4fs"

# Get any artifact
artifact_url = paradime.bolt.get_latest_artifact_url(
    slug=BOLT_SCHEDULE_SLUG, artifact_path="target/catalog.json"
)

​List Bolt schedules

​Get latest runs for a schedule

​Create a Bolt schedule

​Create a minimal schedule

​Create a schedule with notifications, env vars, and self-healing

​Create a schedule using raw dicts (escape hatch)

​Delete a Bolt schedule

​Triggering a Bolt run

​Trigger a run with default commands and branch

​Trigger a run with a custom git branch and PR Number

​Trigger a run with custom commands

​Trigger a run with a custom git branch

​Trigger a run with a reason

​Retrying a Bolt run

​Retry only failed commands by Run ID

​Retry all commands by Run ID

​Streaming logs from a Bolt command

​Stream logs until the command finishes

​One-shot polling with a cursor

​Cancelling a Bolt run

​Getting a Bolt run artifacts

List Bolt schedules

Get latest runs for a schedule

Create a Bolt schedule

Create a minimal schedule

Create a schedule with notifications, env vars, and self-healing

Create a schedule using raw dicts (escape hatch)

Delete a Bolt schedule

Triggering a Bolt run

Trigger a run with default commands and branch

Trigger a run with a custom git branch and PR Number

Trigger a run with custom commands

Trigger a run with a custom git branch

Trigger a run with a reason

Retrying a Bolt run

Retry only failed commands by Run ID

Retry all commands by Run ID

Streaming logs from a Bolt command

Stream logs until the command finishes

One-shot polling with a cursor

Cancelling a Bolt run

Getting a Bolt run artifacts