pyDatorama

What it is

pyDatorama is a Python wrapper around Datorama’s platform API. It models workspaces, streams, and jobs as Python objects and exposes the bulk administrative operations - reruns, renames, metadata collection - that the web UI makes painfully manual.

The problem it solves

Datorama’s interface is built for a handful of pipelines administered interactively. When you administer thousands, clicking through streams is not viable, and the API’s per-day request limits punish naive scripts that re-fetch configuration on every run. The wrapper’s object model plus its restore-from-cache behavior is what made large-scale pipeline automation possible at Ford Direct.

My contribution

I wrote the wrapper and used it in production to automate pipeline management at scale - the same code that powers the Ford Direct case study.

What it covers

• Bulk stream processing, renaming, and rerunning; single-job rerun and stream reprocess
• Metadata collection for all workspaces, streams, and jobs; stream metadata updates
• Object restoration from the last known configuration instead of refetching, keeping request counts inside the daily limit

Engineering decisions

Objects mirror the platform. Workspaces, streams, and jobs are Python objects with the API’s capabilities as methods, so automation scripts read like the domain instead of like HTTP calls.

Cache configuration between runs. The wrapper stores each object’s last definition and rebuilds from it, requesting only new definitions - the difference between a handful of API calls and thousands.

Restore is opt-in per collection. Spaces, streams, and jobs can each be restored from cache or refetched, because sometimes you genuinely need fresh state.

API wrapper example

Illustrative usage with fictional identifiers (no call is made to any vendor from this site):

from Datorama import datorama

# Token issued by Datorama; fictional here
dr = datorama(
    api_token="FICTIONAL-TOKEN",
    verbose=False,
    pause=0.1,
    restore_spaces=False,
    restore_streams=True,   # rebuild stream objects from cached definitions
    restore_jobs=True,
)

# Iterate workspaces and their data streams
for workspace in dr.workspaces.values():
    print(workspace.id, workspace.name)

for stream in dr.streams.values():
    print(stream.id, stream.name)

# A selected stream can rerun all its jobs.
# This is an illustrative call, not executed by the portfolio.
selected_stream = dr.streams.get(12345)
if selected_stream is not None:
    selected_stream.rerun_all_jobs()

Stack

Python, Datorama REST API.

Lifecycle is unspecified on purpose: the wrapper predates this portfolio and I have not verified its current compatibility against Datorama’s API revisions.