Async client

pictograph.AsyncClient is the asyncio counterpart of Client, with the exact same resource surface: every I/O method is a coroutine you await, and every iter(...) accessor returns an async pager you consume with async for. It runs over HTTP/2 on one shared connection pool, so concurrent requests multiplex without threads.

Credential resolution, retries, idempotency keys, typed errors, streaming downloads and the poll helpers are all identical to the sync client.

Source: aio/client.py

Quick start

import asyncio
from pictograph import AsyncClient


async def main() -> None:
    async with AsyncClient(api_key="pk_live_...") as client:
        # await any resource method
        datasets = await client.datasets.list(
            limit=5,
        )

        # async-for the auto-paging iterators
        async for img in client.images.iter(dataset_name=datasets[0].name, directory_path="/train"):
            print(img.filename, img.annotation_count)


asyncio.run(main())

Use it as an async context manager to guarantee socket cleanup, or call await client.aclose() explicitly when you are done.

Concurrent requests

import asyncio
from pictograph import AsyncClient


async def main() -> None:
    async with AsyncClient() as client:
        datasets = await client.datasets.list(
            limit=20,
        )
        # Fetch health insights for every dataset at once.
        reports = await asyncio.gather(
            *(client.datasets.insights(name=d.name) for d in datasets)
        )
        for d, health in zip(datasets, reports):
            print(d.name, health.total_annotations)


asyncio.run(main())

Long-running jobs

The poll helpers are coroutines too. await them, or pass wait=False to fire-and-forget and poll yourself:

async with AsyncClient() as client:
    # Blocks (asynchronously) until the export finishes.
    export = await client.exports.create(
        dataset_name="road-signs",
        name="nightly",
        format="coco",
    )

    # Or start a training run and await completion.
    run = await client.training.create(
        dataset_name="road-signs",
        export_name="nightly",
        pipeline_type="yolox",
        name="detector",
        wait=True,
    )
    print(run.status)

Resources

AsyncClient wires the same resources as Client:

datasets, images, annotations, annotation_comments, exports, training, models, model_evaluations, deployments, credits, notifications, organizations, directories, batch, search, auto_annotate, video, connectors, api_keys, webhooks, workflows.

Method names, arguments and return types match the sync resources one-to-one; only the await differs (aio/resources/). Downloads stream to disk the same way, and iter(...) returns an async pager:

async with AsyncClient() as client:
    # Materialize a full page set, or async-for lazily.
    all_models = await client.models.iter().all()
    async for run in client.training.iter(status="completed"):
        print(run.name)

For CPU-bound local inference (get_model(...).predict(...)) the sync client is still the right tool - ONNX inference does not benefit from asyncio.