Collect run artifacts asynchronously with Polyaxon

An evaluation matrix has finished, and each run has produced a report. You want to bring those reports into a release review without downloading them one at a time or losing track of the runs whose files could not be retrieved.

Polyaxon's AsyncRunClient lets independent artifact downloads overlap inside a Python application. Bound the number of transfers, give each attempt a fresh destination, and retain a collection receipt alongside the files. The result is a local evidence bundle whose completeness you can inspect.

Polyaxon v2.16 added asynchronous artifact and store downloads alongside the async run clients. The async monitoring guide covers status and logs; this walkthrough collects files from an already selected set of runs.

Select the artifacts you actually need

Start from an explicit run inventory. A release review may need outputs/evaluation/report.json from each candidate, while a debugging bundle may need a small directory of predictions and diagnostics. Downloading the entire artifact tree can also collect checkpoints and other large files that the review does not need.

The run client reference distinguishes these operations:

Pass owner, project, and run_uuid together. The artifact path is relative to that run's artifact root; path_to is the local destination root, under which the artifact's relative path is preserved. It is not simply a replacement filename.

Artifact availability and run status are separate observations. A failed run can contain valuable diagnostics, and a successful run can still have a missing or malformed report. Decide which files are required before collecting them. If producers are still writing, wait for their output contract to be complete or use immutable versioned artifacts.

Collect three reports with two download slots

The following example needs Python 3.10 or later, a Polyaxon Python client with the v2.16 async functionality, configured deployment credentials, and network access to the deployment and its artifact-serving compute agents. Replace the organization, project, and UUIDs with runs you can access.

For this example, each producer must have saved a small JSON object at outputs/evaluation/report.json with a run_uuid field identifying that producer run. This is the example's report contract, not a field that Polyaxon automatically inserts into every artifact. The collector accepts at most 5 MiB per report during its post-download content check; it does not enforce a network download-size cap.

Save this as collect-reports.py:

import asyncio
import hashlib
import json
import tempfile
from pathlib import Path

from polyaxon.client import AsyncRunClient

OWNER = "YOUR_ORG"
PROJECT = "YOUR_PROJECT"
RUN_UUIDS = ["FIRST_RUN_UUID", "SECOND_RUN_UUID", "THIRD_RUN_UUID"]
ARTIFACT_PATH = "outputs/evaluation/report.json"
MAX_REPORT_BYTES = 5 * 1024 * 1024


def inspect_report(path, run_uuid):
    if path.stat().st_size > MAX_REPORT_BYTES:
        raise ValueError("Report exceeds the example's inspection limit")
    data = path.read_bytes()
    report = json.loads(data)
    if not isinstance(report, dict) or report.get("run_uuid") != run_uuid:
        raise ValueError("Report does not identify the expected producer run")
    return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}


async def download_and_inspect(run_uuid, staging):
    async with AsyncRunClient(
        owner=OWNER,
        project=PROJECT,
        run_uuid=run_uuid,
        manual_exceptions_handling=True,
    ) as run:
        downloaded = await run.download_artifact(
            path=ARTIFACT_PATH, path_to=str(staging),
        )
    path = Path(downloaded)
    metadata = await asyncio.to_thread(inspect_report, path, run_uuid)
    return path, metadata


async def collect_one(run_uuid, slots, root):
    async with slots:
        staging = Path(tempfile.mkdtemp(prefix="attempt-", dir=root))
        receipt = {
            "owner": OWNER, "project": PROJECT, "run_uuid": run_uuid,
            "artifact_path": ARTIFACT_PATH, "staging": str(staging),
        }
        try:
            source, metadata = await asyncio.wait_for(
                download_and_inspect(run_uuid, staging), timeout=120,
            )
            destination = root / "collected" / run_uuid / "report.json"
            destination.parent.mkdir(parents=True, exist_ok=False)
            source.rename(destination)
            receipt.update(status="collected", local_path=str(destination), **metadata)
        except Exception as exc:
            receipt.update(status="not_collected", error_type=type(exc).__name__)
        return receipt


async def main():
    if len(set(RUN_UUIDS)) != len(RUN_UUIDS):
        raise ValueError("Select each run once")
    root = Path(tempfile.mkdtemp(prefix="polyaxon-reports-", dir=".")).resolve()
    print("Collection directory:", root, flush=True)
    slots = asyncio.Semaphore(2)
    receipts = await asyncio.gather(
        *(collect_one(run_uuid, slots, root) for run_uuid in RUN_UUIDS)
    )
    (root / "collection.json").write_text(json.dumps(receipts, indent=2) + "\n")
    print(json.dumps(receipts, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

Run python collect-reports.py. In a notebook with an existing event loop, call await main() instead of starting another loop. The expected structure is one collection directory with collection.json, a separate staging directory per attempt, and collected/<run_uuid>/report.json for each accepted transfer. The examples have been source-reviewed, not executed against a deployment.

The semaphore allows two active collection attempts. The 120-second deadline starts after a slot is acquired and covers download, client cleanup, and inspection. It requests cancellation rather than imposing a hard wall-clock limit; cleanup can take longer. Python's asyncio documentation describes these cancellation semantics. Canceling a local download does not stop or modify its producer run.

Make partial downloads recognizable

Fresh staging directories matter because the SDK download helper can reuse an existing local path instead of transferring the file again. A path left behind by an interrupted attempt must not become evidence that the next collection succeeded. Passing force=True concerns the remote artifact request; it is not a substitute for choosing a fresh local destination.

The collector moves a file into collected/ only after the download returns and the report passes its content check. Failed attempts retain their staging directories for inspection. A retry should use a new staging directory; do not treat a partially written JSON file as a finished report.

The local SHA-256 identifies the bytes collected. It does not prove they match an independently trusted source digest. If your release contract provides a producer digest or manifest, compare against it before accepting the file. Extend the JSON checks to cover the expected dataset, candidate, evaluator, and schema revisions when those identities matter to the review.

not_collected records an exception type without copying possibly sensitive server error text into the receipt. Investigate the error through your normal diagnostics: a missing file, inaccessible agent, permission failure, timeout, or invalid report calls for different action. The example performs one attempt per run and lets handled failures coexist with successful transfers. Process termination or external cancellation can prevent the final receipt file from being written; a durable collection service should persist receipts incrementally.

Choose streaming and archive behavior deliberately

The async download path writes response chunks to disk, so a file transfer does not require buffering the whole artifact in your application. This example subsequently reads each small JSON report for validation. Large predictions or model weights need a different validator, such as incremental parsing or chunked hashing.

Do not assume that awaiting get_artifact(stream=True) gives your application an async iterator of chunks. The retrieval flag and the Python return shape are different concerns. Use the download helpers when the desired result is a local file.

For a directory, download_artifacts exposes untar, delete_tar, and extract_path. Decide whether the consumer needs an archive or extracted files, keep attempts in separate directories, and inspect the resulting file inventory before declaring the bundle complete. A returned destination string alone does not establish that every expected file is present.

Async I/O overlaps waiting; it does not remove bandwidth, disk, or server limits. Start with a small concurrency bound. For thousands of runs, use a worker queue or bounded batches as well, since a semaphore bounds active work but does not bound the number of tasks created by gather.

Hand off the collection receipt

Treat collection.json as part of the evidence bundle. Compare its run identities with the selected inventory, resolve required not_collected entries, and inspect the reports' own evaluation decisions. A collected file is not a passing candidate.

If collection runs as another Polyaxon operation, retain the receipt and approved report bundle using artifact tracking, along with the source run identities. That gives the review its own run record while preserving where each file came from. The same receipt is useful when collecting locally: it tells the next person which runs were requested, which files arrived, and which gaps still need attention.