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

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

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

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

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.

Returns

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_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 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

none

# 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

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_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

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_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"
)