> For the complete documentation index, see [llms.txt](https://docs.paradime.io/app-help/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.paradime.io/app-help/developers/python-sdk/modules/bolt.md).

# Bolt

## Overview

{% hint style="info" %}

* This feature is available with the [**Paradime Bolt plan**](https://www.paradime.io/pricing)**.**
* Your API keys ***must*** have either [Bolt Schedules Admin or Bolt Schedules Metadata Viewer](/app-help/developers/generate-api-keys.md) capabilities.
  {% endhint %}

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.

{% tabs %}
{% tab title="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.
{% endtab %}

{% tab title="Returns" %}
*`BoltScheduleRuns`*: An object containing the list of Bolt runs with details like ID, status, actor, timestamps, and git information.
{% endtab %}
{% endtabs %}

```python
# 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

{% tabs %}
{% tab title="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*.
{% endtab %}

{% tab title="Returns" %}
*`BoltSchedules`*: An object containing the list of Bolt schedules and the total count of schedules.
{% endtab %}
{% endtabs %}

```python
# 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.

{% tabs %}
{% tab title="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.
{% endtab %}

{% tab title="Returns" %}
*`int`*: The ID of the triggered run.
{% endtab %}
{% endtabs %}

### Trigger a run with default commands and branch

```python
# 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

```python
# 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

```python
# 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

```python
# 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)
```

## Retrying a Bolt run

### Retry the latest failed run of a schedule by Schedule Name

Retries the **latest failed run** of a Bolt schedule by name, 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.

{% tabs %}
{% tab title="Args" %}
**`schedule_name`** *`(str)`*: The name of the schedule whose latest failed run to retry.
{% endtab %}

{% tab title="Returns" %}
*`int`*: The ID of the newly created retry run.
{% endtab %}
{% endtabs %}

```python
# 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 name of the schedule to retry
BOLT_SCHEDULE_NAME = "daily_run"

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

# 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.

{% tabs %}
{% tab title="Args" %}
**`run_id`** *`(int)`*: The ID of the failed run to retry.
{% endtab %}

{% tab title="Returns" %}
*`int`*: The ID of the newly created retry run.
{% endtab %}
{% endtabs %}

```python
# 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 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.

{% tabs %}
{% tab title="Args" %}
**`run_id`** *`(int)`*: The ID of the run to retry.
{% endtab %}

{% tab title="Returns" %}
*`int`*: The ID of the newly created retry run.
{% endtab %}
{% endtabs %}

```python
# 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 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).

{% hint style="info" %}
To get a `command_id`, list the commands for a run with `paradime.bolt.list_run_commands(run_id)`.
{% endhint %}

### Stream logs until the command finishes

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

{% tabs %}
{% tab title="Args" %}
**`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*.
{% endtab %}

{% tab title="Yields" %}
*`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).
{% endtab %}
{% endtabs %}

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

# Trigger a run
run_id = paradime.bolt.trigger_run(schedule_name="dbt run")

# 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.

{% tabs %}
{% tab title="Args" %}
**`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.*
{% endtab %}

{% tab title="Returns" %}
*`BoltCommandLogs`*: An object with `lines` (`List[BoltLogLine]`), `cursor` (str — pass to the next call), and `finished` (bool — flips to `True` once the command exits).
{% endtab %}
{% endtabs %}

```python
# First party modules
import time
from paradime import Paradime

paradime = Paradime(api_endpoint="API_ENDPOINT", api_key="API_KEY", api_secret="API_SECRET")

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.

{% tabs %}
{% tab title="Args" %}
**`run_id`** *`(int)`*: The ID of the run to cancel.
{% endtab %}

{% tab title="Returns" %}
*`none`*
{% endtab %}
{% endtabs %}

```python
# 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.

{% tabs %}
{% tab title="Args" %}
**`schedule_name`** *`(str)`*: The name of the schedule.

**`command_index`** *`(Optional[int])`*: The index of the command in the schedule. Defaults to None.
{% endtab %}

{% tab title="Returns" %}
*`dict`*: The content of the latest manifest JSON.
{% endtab %}
{% endtabs %}

```python
# 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.

{% tabs %}
{% tab title="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.
{% endtab %}

{% tab title="Returns" %}
*`str`*: The URL of the latest artifact.
{% endtab %}
{% endtabs %}

```python
# 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"
)
```


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.paradime.io/app-help/developers/python-sdk/modules/bolt.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.