Async Client

pictograph.AsyncClient is the asyncio counterpart of Client. It exposes 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 a single shared connection pool, so concurrent requests multiplex efficiently — ideal for fanning work out with asyncio.gather.

Credential and configuration resolution is identical to the sync client (explicit kwargs, then PICTOGRAPH_* environment variables, then defaults), and retries, idempotency keys, typed errors, streaming downloads, and the poll helpers all behave the same way.

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(datasets[0].id, folder_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

Because the client multiplexes over one HTTP/2 pool, asyncio.gather runs many calls concurrently without spawning threads:

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(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("road-signs", "nightly", format="coco")

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

Resources

AsyncClient wires the same resources as Client:

datasets, images, annotations, exports, training, models, deployments, credits, organizations, projects, folders, batch, search, auto_annotate, video, connectors, api_keys, webhooks, workflows.

Each is the async twin of the resource documented elsewhere in these docs — the method names, arguments, and return types match one-to-one; only the await differs. Downloads stream to disk exactly as they do on the sync client, 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.