Locomotiv Python SDK

Python access to DataFre8 task discovery, run execution, artifacts, and viewer workflows.

Documentation · PyPI · Issues

The Locomotiv Python SDK is the Python client for the DataFre8/Locomotiv developer-facing platform surface. It communicates through the Intermediate Server public API boundary and does not call the private Cloud Engine directly.

Use it to discover the tasks and capabilities visible to your authenticated account, submit workspace-aware runs, inspect run lifecycle state, work with typed input references, discover produced artifacts, request safe artifact download links, and use optional preview, bridge, and workflow helper surfaces where your account and backend support them.

Available tasks are discovered dynamically. Do not assume a fixed task count or that a task visible in one account, workspace, or environment is visible in another.

What the SDK Provides

• Authentication-aware client setup with environment variables.
• Workspace, product-context, developer-config, and task discovery helpers.
• Task detail and UI-spec inspection for public task contracts.
• DataHub discovery for personal storage, workspace storage, and Public Library rows.
• Typed input references for saved objects and run artifacts.
• Single-input, keyed-input, pair-input, and run-artifact run submission.
• Run polling, cancellation, terminal-state helpers, structured failure fields, and raw diagnostic access.
• Produced-artifact discovery, result-package metadata, and signed download-link retrieval.
• Preview, bridge-session, and limited workflow helpers exposed by the current SDK.

The SDK is not a Cloud Engine client, a frontend SDK, an asset-management admin tool, a billing or team-management SDK, or a substitute for task discovery.

Installation

Install the distribution package:

python -m pip install locomotiv-sdk

Import the SDK package:

from locomotiv_sdk import Locomotiv

For local repository development:

python -m venv .venv
source .venv/bin/activate
python -m pip install -e .

Requirements

Package metadata declares:

• distribution name: locomotiv-sdk
• import name: locomotiv_sdk
• Python requirement: >=3.10
• Python classifiers: 3.10, 3.11, and 3.12
• license metadata: MIT

Configuration

Locomotiv.from_env() reads:

• LOCOMOTIV_BASE_URL
• LOCOMOTIV_API_KEY
• LOCOMOTIV_WORKSPACE_ID

LOCOMOTIV_API_KEY and LOCOMOTIV_BASE_URL are required for from_env(). LOCOMOTIV_WORKSPACE_ID is optional, but workspace-scoped helpers need either an explicit workspace_id argument or an active workspace.

Set the base URL to the public Intermediate Server boundary for your environment. Do not include an API path suffix.

export LOCOMOTIV_BASE_URL="https://api.example.com"
export LOCOMOTIV_API_KEY="replace-with-your-api-key"
export LOCOMOTIV_WORKSPACE_ID="replace-with-your-workspace-id"

Do not commit credentials to source control or notebooks. Rotate credentials if they are exposed.

Connect

from locomotiv_sdk import Locomotiv

loco = Locomotiv.from_env()

if not loco.check_connection():
    raise RuntimeError("Connection check failed.")

profile = loco.myprofile()
print({"connected": True, "profile_loaded": profile.id is not None})

Keep output bounded. Avoid printing raw profile payloads or credentials.

Discover Tasks Dynamically

Use task discovery before choosing a task ID:

from locomotiv_sdk import Locomotiv

loco = Locomotiv.from_env()

tasks = loco.tasks.list()
print({"visible_task_count": len(tasks)})

for task in tasks[:5]:
    print(
        {
            "task_id": task.name,
            "task_kind": task.task_kind,
            "accepted_input_kinds": task.accepted_input_kinds,
        }
    )

Inspect task detail and UI metadata before submitting:

task_id = "replace-with-discovered-task-id"

task_detail = loco.tasks.get(task_id)
ui_spec = loco.tasks.get_ui_spec(task_id)

print(
    {
        "task_id": task_detail.name,
        "task_kind": task_detail.task_kind,
        "has_ui_form": ui_spec.form is not None,
    }
)

Tasks and exposed recipe metadata are related catalog concepts, but they are not interchangeable. Use public task IDs for run submission.

Submit a First Run

Use a typed input reference when the task contract supports an existing saved object:

from locomotiv_sdk import Locomotiv
from locomotiv_sdk.models import UserFileInputRef

loco = Locomotiv.from_env()

run = loco.run(
    task="replace-with-discovered-task-id",
    input_ref=UserFileInputRef("replace-with-user-file-id"),
    workspace_id="replace-with-workspace-id",
    params={},
    publish=True,
)

print({"run_id": run.id, "initial_status": run.status})

Accepted submission is not the same as successful execution. Wait or inspect run detail separately.

final_run = loco.runs.wait(run.id, raise_on_failure=False)

print(
    {
        "status": final_run.status,
        "terminal": final_run.is_terminal,
        "succeeded": final_run.succeeded,
        "failed": final_run.failed,
        "canceled": final_run.canceled,
    }
)

if final_run.failed:
    print(
        {
            "error_code": final_run.error_code,
            "error_detail": final_run.error_detail,
            "request_id": final_run.request_id,
        }
    )

If the backend returns no public structured error, final_run.error may be None. The SDK does not invent failure details.

Typed Inputs and DataHub

DataHub discovery keeps identities separate. A user-file ID, workspace row ID, workspace asset ID, asset-version ID, run ID, and run-artifact identity are not interchangeable.

from locomotiv_sdk import Locomotiv

loco = Locomotiv.from_env()

items = loco.datahub.list_my_storage(limit=25, q="replace-with-search-term")

for item in items[:5]:
    print(
        {
            "scope": item.scope,
            "name": item.name,
            "kind": item.kind,
            "content_type": item.content_type,
            "previewable": item.previewable,
        }
    )

source_ref = items[0].to_user_file_input_ref()
print(source_ref.to_payload())

Strict conversion helpers use explicit identity fields only. They do not infer run inputs from names, filename extensions, MIME types, preview paths, download paths, signed URLs, or asset-version IDs.

For keyed or pair contracts, derive key names from task detail, UI specs, or other public task metadata:

from locomotiv_sdk.models import KeyedInput, RunExecutionOptions, UserFileInputRef

source = KeyedInput(
    key="replace-with-contract-key",
    value=UserFileInputRef("replace-with-user-file-id"),
)

run = loco.run(
    task="replace-with-discovered-task-id",
    inputs=[source],
    workspace_id="replace-with-workspace-id",
    execution=RunExecutionOptions(publish=True),
)

Do not invent input keys or rely on input ordering unless the task contract says so.

Run Status and Artifacts

Successful execution does not guarantee a specific artifact exists. Some tasks produce no artifacts, one artifact, or many artifacts.

run_detail = loco.run_detail("replace-with-run-id")
artifacts = loco.artifacts.from_run_detail(run_detail)

print({"artifact_count": len(artifacts)})

for artifact in artifacts[:5]:
    print(
        {
            "artifact_id": artifact.id,
            "name": artifact.name,
            "type": artifact.type,
            "artifact_type": artifact.artifact_type,
            "runnable": artifact.runnable,
            "downloadable": artifact.downloadable,
        }
    )

When an artifact is downloadable, request a signed download link through the SDK:

artifact = artifacts[0]
link = loco.artifacts.download_link(run_detail.id, artifact.id)

print(
    {
        "download_link_available": bool(link.url),
        "expires_at": link.expires_at,
    }
)

The SDK returns a temporary signed link; it does not write a local file. Treat signed URLs as temporary bearer-style credentials: do not log them, do not store them permanently, do not publish them, and do not commit them.

Runnable artifacts can be passed to a later run only through public run-artifact input metadata:

artifact = artifacts[0]
artifact_ref = artifact.to_runnable_input_ref()
print(artifact_ref.to_payload())

runnable=True is a useful signal, not a guarantee that every downstream task accepts that artifact. Inspect the downstream task contract before chaining.

Preview, Bridge, and Workflow Helpers

The SDK exposes optional advanced surfaces through:

• loco.preview for backend-assisted previews from saved objects with preview metadata
• loco.bridge for controlled viewer handoff sessions
• loco.workflows for the limited public workflow helper surface

These helpers depend on account, workspace, entitlement, object metadata, and backend capability. Do not log preview access metadata, bridge launch URLs, or token-like values.

Documentation

The complete Python SDK Help Center manual contains 33 articles covering setup, task discovery, DataHub, inputs, runs, artifacts, preview, bridge sessions, workflow helpers, and troubleshooting.

Read the public Help Center documentation at:

https://lantern.datafre8.com/help

Development

Run the unit suite from the repository root:

.venv/bin/python -m unittest discover -s tests/unit -p "test*.py" -t .

Public examples are available under examples/. Example scripts use placeholders and environment configuration; review each script before running it against a real account.

Safety Notes

• Do not print or commit API keys.
• Do not include authorization headers in tickets or logs.
• Do not construct internal URLs or storage paths.
• Do not call Cloud Engine directly.
• Do not assume task counts, task IDs, workspace IDs, or artifact identities are stable across accounts.
• Do not treat raw payloads as the preferred stable SDK contract.
• Do not treat a local filesystem path as a server-side run input unless you intentionally use an upload or snapshot flow.

License and Links

Package metadata declares the license as MIT.

Repository metadata:

• Homepage: https://github.com/datafre8-company/locomotiv-sdk
• Documentation: https://lantern.datafre8.com/help
• Repository: https://github.com/datafre8-company/locomotiv-sdk
• Issues: https://github.com/datafre8-company/locomotiv-sdk/issues