Bolt

Overview

• This feature is available with the Paradime Bolt plan.

• Your API keys must have either Bolt Schedules Admin or Bolt Schedules Metadata Viewer capabilities.

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.

List Bolt schedules

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

Args

schedule_name (str): The name of the Bolt schedule. Must be exact schedule name.

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.

Returns

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime( api_endpoint="API_ENDPOINT", api_key="API_KEY", api_secret="API_SECRET")

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

Get latest runs for a schedule

Args

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.

Returns

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime(
    api_endpoint="API_ENDPOINT",
    api_key="API_KEY",
    api_secret="API_SECRET"
)

# Get the latest 10 runs for a schedule
recent_runs = paradime.bolt.list_runs(
    schedule_name="daily_run",
    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(
    schedule_name="daily_run",
    offset=10,
    limit=10
)

# Get all recent runs (up to 100)
all_recent_runs = paradime.bolt.list_runs(
    schedule_name="daily_run",
    limit=100
)

Triggering a Bolt run

Triggers a run for a given schedule name.

Args

schedule_name (str): The name of the schedule to trigger the run for.

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.

Returns

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_key="API_KEY", api_secret="API_SECRET")

# Replace with the Bolt schedule name to trigger
BOLT_SCHEDULE_NAME = "daily_run"

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

# 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_key="API_KEY", api_secret="API_SECRET")

# Replace with the Bolt schedule name to trigger
BOLT_SCHEDULE_NAME = "daily_run"

# 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(schedule_name=BOLT_SCHEDULE_NAME, 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_key="API_KEY", api_secret="API_SECRET")

# Replace with the Bolt schedule name to trigger
BOLT_SCHEDULE_NAME = "daily_run"

# 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(schedule_name=BOLT_SCHEDULE_NAME, 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_key="API_KEY", api_secret="API_SECRET")

# Replace with the Bolt schedule name to trigger
BOLT_SCHEDULE_NAME = "daily_run"  

# 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(schedule_name=BOLT_SCHEDULE_NAME, branch=GIT_BRANCH)

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

Cancelling a Bolt run

Cancels a Bolt run.

Args

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

Returns

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime( api_endpoint="API_ENDPOINT", api_key="API_KEY", api_secret="API_SECRET")

# 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

schedule_name (str): The name of the schedule.

command_index (Optional[int]): The index of the command in the schedule. Defaults to None.

Returns

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime( api_endpoint="API_ENDPOINT", api_key="API_KEY", api_secret="API_SECRET")

# Replace with the Bolt schedule name you want to get artifacts from
BOLT_SCHEDULE_NAME = "daily_run" 

# Get manifest.json dictionary
manifest_json = paradime.bolt.get_latest_manifest_json(schedule_name=BOLT_SCHEDULE_NAME)

Get latest run artifacts URL

Retrieves the URL of the latest artifact for a given schedule.

Args

schedule_name (str): The name of the schedule.

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.

Returns

# First party modules
from paradime import Paradime

# Create a Paradime client with your API credentials
paradime = Paradime( api_endpoint="API_ENDPOINT", api_key="API_KEY", api_secret="API_SECRET")

# Replace with the Bolt schedule name you want to get artifacts from
BOLT_SCHEDULE_NAME = "daily_run" 

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