Dataerai Python SDK

Python clients for the Dataerai transfer daemon and the complete public REST API. The daemon client supports authenticated uploads, downloads, metadata, and resumable transfers; the optional REST client exposes every public API path.

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[rest]>=0.2.0b1,<0.3"        # Full public REST API
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
pip install --pre "dataerai-sdk[envelope]>=0.2.0b1,<0.3"    # Encrypted-envelope reader

Full public REST API

The daemon client remains the best path for resumable file transfers. Install the rest extra when a script also needs any public console API endpoint:

from dataerai.rest import RestClient

# Reuses credentials written by `dataerai auth login --device`.
with RestClient() as api:
    response = api.request(
        "GET",
        "/api/projects/",
        params={"page_size": 20},
    )
    response.raise_for_status()
    projects = response.json()

request() accepts every relative /api/ path, including endpoints added after this SDK version was published, and returns the raw httpx.Response. stream() applies the same security checks while iterating a large response. Absolute URLs and redirects are rejected so bearer credentials remain scoped to the configured Dataerai server. For headless jobs, set DATAERAI_SERVER and DATAERAI_TOKEN, or provide the CLI-compatible JSON credential object through DATAERAI_CREDENTIALS_JSON.

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. Every checkpoint is a first-class model asset, so provenance graphs render it as a dedicated violet compute node with the Model brain-circuit logo. Final checkpoints may also publish hashed figures as analysis assets linked to the model with visualizes relationships. When the notebook session is tracing, the tracker automatically reuses its run ID across models, manifests, figures, and the execution log.

For notebook runs that publish many checkpoints or large analysis artifacts, %dataerai --request-timeout 120 ... raises the daemon request/response limit without changing the separate upload-transfer timeout.

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=5,
    callbacks=[tracker.callback(training_parameters={"batch_size": 128})],
)
final = tracker.save_checkpoint(
    epoch=5,
    label="final",
    record_name="mnist-mlp",
    metrics={"test_accuracy": 0.91},
    training_parameters={"batch_size": 128, "epochs": 5},
    outcomes={"sample_predictions": sample_predictions},
    analysis_artifacts=[
        {
            "path": "artifacts/training-curves.png",
            "kind": "training-curves",
            "caption": "Training and validation metrics across five epochs.",
            "media_type": "image/png",
        }
    ],
)

PyTorch neural-network provenance

Install the PyTorch adapter when a training run must preserve native model and optimizer state:

pip install "dataerai-sdk[nn-pytorch]"

PyTorchProvenanceTracker emits the same DataFed_TorchFlow preservation concepts as the TensorFlow tracker while keeping framework-native checkpoint validation inside the adapter. Each .pt file contains only the versioned schema, run/epoch identity, model state_dict, and optimizer state_dict. Before any upload, the tracker reloads the file on CPU with weights_only=True; a checkpoint that cannot pass that restricted loader is rejected.

import torch

from dataerai.nn import PyTorchProvenanceTracker

model = MyModel()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
tracker = PyTorchProvenanceTracker(
    model,
    optimizer,
    session=dataerai_session,
    run_name="mnist-cnn",
    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,
)

for epoch in range(1, 6):
    train_one_epoch(model, optimizer)
    tracker.save_checkpoint(
        epoch=epoch,
        label="epoch",
        metrics={"loss": current_loss},
        training_parameters={"learning_rate": 1e-3, "epochs": 5},
    )

With a NotebookSession, checkpoints are published as model assets. Manifests and optional analysis figures are ordinary analysis assets. Qualified derived_from, describes, and visualizes relationships connect them to datasets, the training notebook, and the preceding checkpoint. Without a session, the identical checkpoint and manifest contracts are written locally.

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

API reference

Encrypted envelopes

The optional reader keeps OAuth credentials in the Go daemon. Python receives only the audited per-envelope DEK over the owner-only local IPC endpoint:

from dataerai import envelope

with envelope.open("data.denv") as env:
    print(env.audit)
    env.verify_signature()  # server-bound cache; offline while fresh (5 minutes)
    env.unlock()            # daemon IPC; audited purpose="open"
    print(env.namelist())
    with env.open("member.csv") as member:
        first_kib = member.read(1024)
    env.extractall("out")   # atomic; failures leave no partial output

Populate or refresh the public-key cache with dataerai envelope verify data.denv. A new Python process must call unlock() again; neither the SDK nor daemon persists the DEK.

The whole path — parsing, fresh-cache verification, unlock() and extractall() — is portable. The daemon transport is an AF_UNIX socket on POSIX and a named pipe on Windows, so no platform needs to fall back to the Go CLI.

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

Choosing an allocation

upload() and update_content() accept an allocation_id; this is how you get one:

for a in client.list_allocations(project_id):
    print(a.allocation_id, a.alias, a.repository_name,
          f"{a.bytes_free:,} B free", f"{a.records_free:,} records free",
          "FULL" if a.is_full else "")

Checking before you upload is worth the round trip. A full allocation doesn't fail at the point of cause — uploads are rejected, and a rejected upload can leave a contentless asset that surfaces much later as "no downloadable content".

is_full is necessary, not sufficient. True means an upload will be rejected. False does not guarantee one succeeds: the console also hard-blocks on billing suspension, enforces a temporary grace ceiling instead of data_volume_limit during a grace window, and walks up parent allocations — none of which the daemon reports, so none are visible here. Read False as "the reported quota has room".

A limit of 0 means zero capacity, not unlimited — the console's gate is a plain current + incoming > limit with no special case for zero. A missing limit likewise reads as full rather than unlimited: fail-closed is the safe direction, because wrongly reporting room is the failure this is meant to prevent.

An empty list does not mean uploads will fail. The server resolves an upload's allocation through a cascade — the owner's allocation on its preferred repository, else the owner's default, else, for a project-owned asset, one held by the project's owning user. That last step is invisible here, because the daemon lists only what the project holds directly. So an upload with no allocation_id may land in an allocation that is not in this list; passing one from here pins the destination instead.

Listing what you have

list_collection_assets() / find_collection_assets() enumerate a collection's contents without downloading them — the alternative before they existed was download_collection(), which answers the question by fetching everything:

for asset in client.find_collection_assets(collection_id):
    print(asset.asset_id, asset.title, asset.size_bytes, asset.has_content)

find_collection_assets() pages until exhausted, and raises rather than looping if the server claims another page without a cursor or repeats one.

list_projects() and list_collections() show what you can upload into, not everything you can read. Both are projections of list_tree(), and the daemon omits any project you lack WRITE_METADATA on — so a project you can read but not write to is absent, and so are its collections. There is no SDK call that lists read-only projects today.

list_projects() populates only project_id, name and root_collection_id; tree.list carries nothing else, so the remaining Project fields are None. Collections shared with you directly are attached to the personal node, so one returned under owner_type="user" may carry another project's owner_project_id.

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.

What a collection actually contains

list_collection_assets() enumerates assets. get_collection_manifest() enumerates the files inside them, flattened, with each one's position in the collection tree — and, crucially, which of them a download or sync will silently not deliver:

manifest = client.get_collection_manifest(collection_id)
print(f"{manifest.total_files} files, {manifest.total_bytes:,} bytes")

for entry in manifest.undownloadable:
    why = "external" if entry.external else "no checksum on the server"
    print(f"  will NOT arrive: {entry.relative_path}  ({why})")

This is the answer to "why does my synced folder have fewer files than the collection?". A sync omits two kinds of entry and reports neither per-file: external entries, whose bytes live outside Dataerai, and entries the server holds no checksum for, which the daemon refuses to write because it cannot verify them (the skipped_unverified count). The daemon collapses both into one is_downloadable flag, so external is what tells them apart.

The manifest is complete or it raises — the daemon assembles every page itself and rejects a result whose totals or snapshot shifted while paging, which is what makes it safe to use as the expectation you check a sync against. Note it carries no checksums; the checksum is consumed to compute is_downloadable, not forwarded.

collection_id must be a canonical UUID here — lower-case, hyphenated, no braces or urn: prefix. The daemon rejects other spellings rather than normalising them.

download_collection() raises on a failed or timed-out transfer, but an asset the daemon could never queue is reported rather than raised: its ID lands in result.failed_assets, and it is absent from both result.transfers and result.asset_count. Check that field before treating the download as complete.

result = client.download_collection(collection_id, "/data/out")
if result.failed_assets:
    raise RuntimeError(f"{len(result.failed_assets)} assets did not download: "
                       f"{result.failed_assets}")

Watching and controlling transfers

The daemon runs a real job engine behind upload() and download(): chunked multipart, parts in parallel, resume-from-partial across restarts, and at most four transfers at once with the rest queued. These expose it:

for t in client.list_transfers():
    print(t.transfer_id, t.status, t.direction, f"{t.percent:.0f}%")

# Transfers started anywhere — the web app, another device, a runner
page = client.list_server_transfers(type="upload")
print(page.count, "total;", [t.executor for t in page.transfers])

pause_transfer() and resume_transfer() are fire-and-forget, and therefore silent on failure. The daemon writes nothing on success, so they cannot wait for an acknowledgement — and one naming an unknown transfer is discarded without raising. This mirrors the Node SDK rather than inventing a Python-only contract. Confirmation is asynchronous: the worker emits transfer.paused, and list_transfers() then shows paused.

cancel_transfer() is different — the daemon acks it, so it waits and raises ERR_TRANSFER_NOT_FOUND if the transfer is unknown.

Stopping the daemon

shutdown_daemon() completes the set of fire-and-forget commands the Node SDK sends. The daemon logs the request, signals its own shutdown and closes the connection without replying.

⚠️ The daemon is shared machine-wide. Stopping it aborts in-flight transfers for every client — the desktop app and any running CLI included. Use close() unless you specifically mean to stop the daemon, e.g. tearing down an ephemeral environment or a test fixture.

The client is closed afterwards: one left open against a stopped daemon fails every later call with a connection error that says nothing about the cause. Build a new client to reconnect — one with auto_start and a binary_path respawns the daemon.

What a blocking call sees. Cancelling a transfer that an upload() or download() is waiting on makes that call raise DaemonError with code == "cancelled" — the daemon emits transfer.failed with that code rather than a transfer.cancelled message, so check the code rather than expecting a clean return. Pausing is different and easier to get wrong: a paused transfer sends no terminal event, so a blocked upload() keeps waiting and eventually raises DaemonTimeoutError after its transfer_timeout_s. Pausing does not extend that budget.

Cancelling an upload that is still running needs its transfer id, and upload() does not return until it finishes — so capture the id from a progress event:

seen = {}
client.upload(path, ..., on_progress=lambda e: seen.setdefault("id", e.transfer_id))
# ... from another thread, once seen has an id:
client.cancel_transfer(seen["id"])

Keeping a local folder in sync

download_collection() fetches everything, every time. sync_collection() compares the collection against what is already on disk and moves only what changed, keeping state under the destination so it works across runs:

result = client.sync_collection(
    collection_id,
    "/data/my-collection",
    on_progress=lambda p: print(f"  {p.percent:.0f}%  {p.path}"),
)
print(f"{result.downloaded} downloaded, {result.pruned} removed, "
      f"{result.elapsed_ms} ms")

if not result.is_complete:
    print(
        f"{result.failed} files could not be fetched; "
        f"{result.conflicts} local conflicts were preserved"
    )
if result.skipped_unverified:
    print(f"{result.skipped_unverified} withheld — the server had no checksum")

⚠️ prune=True deletes local files. Anything under the destination the collection no longer contains is removed. Preview it first — dry_run=True computes the same plan, writes nothing, deletes nothing, and does not even create a missing destination:

preview = client.sync_collection(cid, dest, prune=True, dry_run=True)
print(f"{preview.plan.prune_candidates} local files would be deleted")
print(f"{preview.plan.to_download} would be fetched ({preview.plan.bytes:,} bytes)")

A partial sync does not raise. Files that could not be fetched are counted in failed. Locally modified files that the daemon preserved are counted in conflicts. Either makes is_complete false because the destination was not brought fully in line. Files that were never going to arrive are counted separately in skipped_external and skipped_unverified, and deliberately do not affect is_complete, since the plan declared them up front. get_collection_manifest() names them individually.

A sync cannot be cancelled. The daemon runs it detached from the request, so sync_timeout_s bounds how long you wait — not the sync, which keeps running and keeps writing. Closing the client also abandons only your wait if the shared daemon remains alive. Only one sync may be active per destination; a second raises DaemonError.

Updating an existing asset

Metadata and content are updated by two different calls, both taking an asset_id:

client.set_metadata(asset_id, tags=["v2"])              # metadata
client.update_content(asset_id, "v2.csv")               # a new content version

Earlier versions are kept. List them and fetch an older one by ID:

for v in client.list_content_versions(asset_id):
    print(v.content_id, v.status, v.size_bytes, v.created_at)

client.download(asset_id, "out/", content_id="<older content_id>")

A version appears in that list as soon as its upload starts, so an in-flight one shows up with status="uploading" and is_downloadable == False — it has no bytes behind it yet. They are surfaced rather than hidden so a concurrent upload doesn't look like nothing is happening.

Order: in-flight versions come first, then the available ones newest-first. So [0] is not reliably the newest downloadable version while an upload is running — filter rather than index:

newest = next(v for v in client.list_content_versions(asset_id)
              if v.is_downloadable)

upload() with a title that already exists updates the content but silently drops the metadata. The daemon upserts by (collection, title, owner); when that matches, the console returns the existing asset unchanged and attaches a new content version. The bytes land, but description, alias, record_type, tags and metadata from that call are ignored with no error. Use update_content() to add a version to a known asset, and set_metadata() to change fields — both take an asset_id, so neither can hit the wrong asset.

Bulk operations

upload() takes a sequence of paths to build one asset from several files — the daemon puts them in a single AssetContent and moves them under one transfer. Filenames must be unique within the asset, since the filename is the server-side object key:

client.upload(["run.csv", "run.json"], title="Run 7",
              owner_type="project", owner_id=project_id)

upload_many() is the other axis: one asset per item, uploaded with bounded concurrency. The keyword arguments are batch defaults and every UploadItem field overrides the default of the same name, so assets that differ in only a field or two stay readable:

from dataerai import UploadItem

result = client.upload_many(
    [
        UploadItem("s1.csv", title="Sample 1", metadata={"well": "A1"}),
        UploadItem("s2.csv", title="Sample 2", metadata={"well": "B2"}),
        UploadItem(["s3.csv", "s3.json"], title="Sample 3"),
    ],
    owner_type="project",
    owner_id=project_id,
    record_type="dataset",            # applies to all three
    metadata={"run": "2026-07-28"},   # merged into each item's own metadata
    concurrency=4,
)
result.raise_for_failures()

metadata is shallow-merged — batch keys first, then the item's, item wins on collision. Every other field (including tags) replaces the default outright, so a shared tag can be dropped for one item.

set_metadata_many() does the same for metadata-only edits:

from dataerai import MetadataUpdate

client.set_metadata_many([
    MetadataUpdate(a_id, tags=["qc-pass"], metadata={"well": "A1"}),
    MetadataUpdate(b_id, tags=["qc-fail"], metadata={"well": "B2"}),
]).raise_for_failures()

Note this is client-side fan-out, not a protocol batch: the daemon exposes only a single-asset asset.metadata.set, so it issues one request per asset. You get bounded concurrency, one call, and one failure report — not fewer round-trips.

Both bulk calls finish the batch instead of aborting on the first failure. Aborting halfway through a large batch leaves you worse off than finishing and reporting, so failures land in result.failed (each with the item's input index) and successes in result.succeeded. A caller that ignores failed will read a partial batch as a complete one — check result.ok or call result.raise_for_failures(), which raises BulkOperationError.

result.results is positionally aligned with the input list — results[i] is the outcome of item i, or None if it failed. That is how you map an asset back to the item that produced it, which matters precisely because each item carries its own metadata:

for item, uploaded in zip(items, result.results):
    if uploaded is not None:
        print(item.title, "->", uploaded.asset_id)

succeeded and failed are convenience views over the same outcomes, both in input order rather than completion order.

On concurrency: raising it past the default of 4 does not buy throughput. The daemon runs at most 4 transfers at once and queues the rest (defaultMaxConcurrent in cli/internal/transfer/manager.go, with no configuration knob), so a higher value only parks more SDK worker threads on transfers the daemon has not started. Note also that transfer_timeout_s starts when the daemon accepts a transfer, not when it starts moving bytes — time spent queued counts against it, so a large batch with a lowered timeout can fail items purely from queueing.

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

Daemon events

on_progress= reports on one call's own transfer. client.on() subscribes to the daemon itself, which is the only way to observe a collection sync, a non-fatal transfer.error, or a pause.

client = DataeraiClient()
client.connect()

@client.on("transfer.error")
def on_error(evt):
    # Not a failure: when evt.retrying is set the daemon retries by itself.
    print(f"{evt.transfer_id}: {evt.message} (retrying={evt.retrying})")

@client.on("collection.sync_complete")
def on_sync(evt):
    if evt.skipped_unverified:
        print(f"{evt.skipped_unverified} files skipped — no checksum offered")

The daemon broadcasts to every connected client, so handlers also see work started by the desktop app or the CLI, not just this process.

Handlers run on the reader thread and must return promptly. The daemon buffers 64 events per connection and drops the overflow rather than blocking other clients — a slow handler loses events outright, and they are not redelivered. Queue the work; don't do it in the handler.

In particular, do not call a blocking client method from a handler. The reader thread is what delivers responses, so such a call waits for itself and stalls until its request timeout. Put the id on a queue and act on it from your own thread.

An unknown event name raises ValueError at registration rather than never firing. dataerai.EVENT_NAMES holds the full set.

Subscribe to "*" to see everything — for logging, or to find out what the daemon actually emits during an operation. A wildcard handler gets the same parsed event object, so use event_name() when the name matters:

from dataerai import event_name

client.on("*", lambda e: print(event_name(e), e))

Handlers registered for an event's own name run before wildcard handlers.

Exceptions raised inside a handler never reach the reader thread, but they are logged to the dataerai.events logger rather than discarded — a handler with the wrong signature raises on every event, and a silent version of that is indistinguishable from the daemon never sending one.

Each distinct failure is reported once, with its traceback. A broken handler fails on every matching event and transfer.progress arrives continuously, so reporting every occurrence would bury the first traceback under thousands of copies. Two different broken handlers still get one report each. Silence the lot with logging.getLogger("dataerai.events").setLevel(logging.CRITICAL).

Watching an upload() or download() finish? Subscribe to asset.upload_complete / asset.download_complete, not transfer.complete. For an asset transfer the daemon sends the asset.* event instead of the generic one — never both — so transfer.complete stays silent for exactly the transfers the SDK starts. The same substitution applies to transfer.failed.

Names are the daemon's own wire names, so the string here is the string in the protocol reference and in the daemon's logs. Porting from the Node SDK, which spells them transfer:fileComplete: drop the camelCase and use . for :.

Error types