Dataerai Python SDK

Python client for the Dataerai transfer daemon.
Supports authenticated uploads, downloads, and metadata operations. The daemon is started automatically if it is not already running.

Requirements

• Python ≥ 3.10
• The dataerai binary must be installed and on PATH (or pass binary_path explicitly)
• The user must be logged in via dataerai auth login before calling auth_status() / upload() / download()

Installation

pip install --pre "dataerai-sdk>=0.2.0b1,<0.3"
# or, from source:
pip install -e sdk/python/

Optional ML adapters are installed separately so the base SDK stays small:

pip install --pre "dataerai-sdk[ml]>=0.2.0b1,<0.3"          # PyTorch Zarr datasets
pip install --pre "dataerai-sdk[ml-keras]>=0.2.0b1,<0.3"    # Keras batches
pip install --pre "dataerai-sdk[ml-hls4ml]>=0.2.0b1,<0.3"   # HLS4ML conversion
pip install --pre "dataerai-sdk[notebook]>=0.2.0b1,<0.3"    # IPython %dataerai magic
pip install --pre "dataerai-sdk[nn-tensorflow]>=0.2.0b1,<0.3" # TensorFlow provenance

Quick start

from dataerai import DataeraiClient

with DataeraiClient(binary_path="/usr/local/bin/dataerai") as client:
    # Check auth
    status = client.auth_status()
    print(f"Logged in as {status.user_email} ({status.user_id}), token expires {status.expires_at}")
    if status.user_id is None:
        raise RuntimeError("Upgrade the dataerai CLI to use user-owned uploads")

    # Upload a user-owned file. owner_id is a UUID, not an email address.
    result = client.upload(
        "/path/to/data.csv",
        title="My dataset",
        owner_type="user",
        owner_id=status.user_id,
        record_type="dataset",
        on_progress=lambda p: print(f"  {p.percent:.0f}%  {p.rate_mbps:.1f} MB/s"),
    )
    print(f"Uploaded  asset_id={result.asset_id}  content_id={result.content_id}")

    # Download it back
    dl = client.download(result.asset_id, dest_dir="/tmp/downloads")
    for f in dl.files:
        print(f"  {f.local_path}  ({f.size:,} bytes)")

    # Read / update metadata
    meta = client.get_metadata(result.asset_id)
    updated = client.set_metadata(
        result.asset_id,
        title="My dataset v2",
        record_type="analysis",
        tags=["csv", "demo"],
    )

    # Link a derived result to the source asset that produced it
    relationship = client.create_relationship(
        result.asset_id,
        "source-asset-id",
        relationship_type="derived_from",
        analysis_mode="non_destructive",
        qualifiers={"tool": "pycroscopy"},
    )
    related_id = relationship.related_asset["id"] if relationship.related_asset else "source-asset-id"
    print(f"Linked via {relationship.type} to {related_id}")

Notebook magic

After signing in with the Dataerai CLI, a notebook only needs a destination path. The first component names the project; the notebook magic creates that project when it is missing, then creates any missing collection components beneath it:

DESTINATION_COLLECTION_PATH = "Research / Experiments / July"

%load_ext dataerai.magics
%dataerai $DESTINATION_COLLECTION_PATH

The magic publishes a dataerai_session variable. Its uploads automatically target the selected project and collection:

dataset = dataerai_session.find_asset(
    "did:dataerai:beta:asset:...",
    title="input.csv",
)
dataerai_session.download(dataset, "source-data")
result = dataerai_session.upload(
    "analysis.csv",
    metadata={"component": "derived-output"},
)

Use %dataerai --as run $DESTINATION_COLLECTION_PATH to choose a different session variable. Regular Python code can call connect_notebook(DESTINATION_COLLECTION_PATH) instead.

Trace a notebook run

Add --trace to capture every subsequent cell's source, timestamps, stdout, stderr, structured Python logging records, rich display outputs, returned value, and error. Uploads and downloads through the session join the same run automatically, following the run-centered provenance pattern used by the QICK integration:

%dataerai --trace --notebook analysis.ipynb --title "July analysis" $DESTINATION_COLLECTION_PATH

# Run analysis cells and upload their products through dataerai_session.
result = dataerai_session.upload(
    "analysis.csv",
    record_type="analysis",
    metadata={"component": "derived-output"},
)

%dataerai --finish

%dataerai --finish waits until its cell completes, then uploads a JSON execution-log asset with record_type="log". Every uploaded asset carries a shared notebook_run_id plus dataerai-notebook-trace and notebook-run:<run-id> tags. The log is linked to each recorded input and product with records_telemetry, and the published result is available as dataerai_trace. Run identity is merged onto an existing same-title asset, so re-running a fixed notebook or output filename keeps both run tags searchable. If a traced cell fails, its source, outputs, error, and traceback are published immediately; an explicit %dataerai --finish is not required for that failed run.

Tracing is deliberately opt-in because cell source and output can contain secrets or sensitive research data. Review the notebook before enabling it. The recorder does not read environment-variable values or scan/upload arbitrary files; a file becomes a recorded product only when the notebook uploads it through the traced session. Finish an active trace before starting another one in the same Python kernel.

TensorFlow neural-network provenance

TensorFlowProvenanceTracker implements the complete preservation contract ported from DataFed_TorchFlow. It saves native .keras checkpoints containing model and optimizer state, writes a machine-readable manifest, captures architecture and hyperparameters, data and training-code checksums, outcomes, and full runtime/system properties, and authors derived-from links to datasets, the notebook, and prior checkpoints.

from dataerai.nn import TensorFlowProvenanceTracker

tracker = TensorFlowProvenanceTracker(
    model,
    session=dataerai_session,
    run_name="mnist-mlp",
    dataset_references=[
        {
            "asset_id": dataset_asset.asset_id,
            "source_asset_did": SOURCE_ASSET_DID,
            "filename": DATA_PATH.name,
            "sha256": dataset_sha,
        }
    ],
    notebook_path=NOTEBOOK_PATH,
    notebook_asset_id=notebook_asset.asset_id,
)
history = model.fit(
    x_train,
    y_train,
    epochs=2,
    callbacks=[tracker.callback(training_parameters={"batch_size": 128})],
)
final = tracker.save_checkpoint(
    epoch=2,
    label="final",
    record_name="mnist-mlp",
    metrics={"test_accuracy": 0.91},
    training_parameters={"batch_size": 128, "epochs": 2},
)

System capture intentionally does not enumerate environment variables. It records OS/host, CPU and memory properties, TensorFlow-visible devices, optional NVML GPU memory, Python/framework/package versions, and only a small allowlist of determinism and thread settings.

API reference

DataeraiClient(*, socket_path, binary_path, auto_start, start_timeout_s, request_timeout_s)

Use as a context manager (with DataeraiClient(...) as client:) for automatic cleanup, or call client.connect() / client.close() manually.

Methods

For headless notebooks, pass record_type to upload() or set_metadata(). Both operations use the authenticated transfer daemon, so integrations do not need to read the CLI credential store or expose access tokens to Python.

ML adapters

The dataerai.ml package reads Dataerai-hosted Zarr image stores directly from the short-lived /zarr-access/ bundle. PyTorch users can keep using ZarrAssetDataset / ZarrOriginalsDataset; Keras users get matching batch datasets:

from dataerai.ml import access
from dataerai.ml.keras_dataset import ZarrOriginalsSequence

creds = access.load_credentials()
train = ZarrOriginalsSequence(
    creds,
    asset_id="dataset-asset-id",
    size=64,
    batch_size=32,
    shuffle=True,
    label_mode="categorical",
)

model.fit(train, epochs=5)

For HLS4ML conversion, keep inputs in Keras' channels-last layout:

from dataerai.ml.hls4ml import convert_keras_model

hls_model = convert_keras_model(
    model,
    output_dir="hls-project",
    project_name="dataerai_model",
    data=train,              # writes input/output .npy testbench arrays
    testbench_batches=2,
)

create_relationship() arguments

Authors a directed provenance edge from_asset_id → to_asset_id. You need write access to the source and read access to the target. rel_type is a free-form verb describing the source's role, e.g. "analysis_of" or "acquired_with". Notebook integrations can pass the same value with the keyword-only alias relationship_type.

# Link a processed result back to the raw data it came from.
client.create_relationship(analysis.asset_id, raw.asset_id, "analysis_of",
                           analysis_mode="non_destructive")

upload() keyword arguments

Progress events

@dataclass
class ProgressEvent:
    transfer_id: str
    bytes_done: int
    bytes_total: int
    chunk_index: int
    chunk_count: int
    rate_bps: float
    file_index: int
    file_name: str

    @property
    def percent(self) -> float: ...   # 0–100

    @property
    def rate_mbps(self) -> float: ...

Error types