Async support

Why use async?

When working with external datasets, such as Tilebox datasets, loading data may take some time. To speed up this process, you can run requests in parallel. While you can use multi-threading or multi-processing, which can be complex, often times a simpler option is to perform data loading tasks asynchronously using coroutines and asyncio.

Switching to an async datasets client

To switch to the async client, change the import statement for the Client. The example below illustrates this change.

from tilebox.datasets import Client

# This client is synchronous
client = Client()

from tilebox.datasets.aio import Client

# This client is asynchronous
client = Client()

After switching to the async client, use await for operations that interact with the Tilebox API.

# Listing datasets
datasets = client.datasets()

# Listing collections
dataset = datasets.open_data.copernicus.sentinel1_sar
collections = dataset.collections()

# Collection information
collection = collections["S1A_IW_RAW__0S"]
info = collection.info()
print(f"Data for My-collection is available for {info.availability}")

# Loading data
data = collection.query(temporal_extent=("2022-05-01", "2022-06-01"), show_progress=True)

# Finding a specific datapoint
datapoint_uuid = "01910b3c-8552-7671-3345-b902cc0813f3"
datapoint = collection.find(datapoint_uuid)

# Listing datasets
datasets = await client.datasets()

# Listing collections
dataset = datasets.open_data.copernicus.sentinel1_sar
collections = await dataset.collections()

# Collection information
collection = collections["S1A_IW_RAW__0S"]
info = await collection.info()
print(f"Data for My-collection is available for {info.availability}")

# Loading data
data = await collection.query(temporal_extent=("2022-05-01", "2022-06-01"), show_progress=True)

# Finding a specific datapoint
datapoint_uuid = "01910b3c-8552-7671-3345-b902cc0813f3"
datapoint = await collection.find(datapoint_uuid)

Accessing assets asynchronously

The storage client is asynchronous. Resolve the assets from one queried datapoint, then await the storage operation:

from tilebox.datasets.assets import AssetCollection
from tilebox.storage.aio import Client as StorageClient

datasets = await client.datasets()
collections = await datasets.open_data.aws_earth.sentinel2.collections()
data = await collections["L2A"].query(temporal_extent=("2025-01-01", "2025-01-02"))

assets = AssetCollection.from_datapoint(data.isel(time=0))
storage = StorageClient()
contents = await storage.read_bytes(assets["thumbnail"], max_bytes=10_000_000)

See Read and download assets for streaming, downloads, and GeoTIFF window reads.

Downloading assets concurrently

Run independent storage operations concurrently with asyncio.gather. This example continues from the preceding query and downloads the datapoint’s red, green, and blue bands:

import asyncio

await asyncio.gather(
    storage.download(assets["red"], "red.tif"),
    storage.download(assets["green"], "green.tif"),
    storage.download(assets["blue"], "blue.tif"),
)

Async workflows

Python workflow tasks can define execute with async def. The runner waits for the method to complete, so you can await asynchronous APIs such as Tilebox Storage directly without wrapping the task code in asyncio.run().

For example, a task can read a small group of assets concurrently:

import asyncio

from tilebox.datasets.assets import Asset
from tilebox.storage.aio import Client as StorageClient
from tilebox.workflows import ExecutionContext, Task

storage = StorageClient()

class ReadAssets(Task):
    assets: list[Asset]

    async def execute(self, context: ExecutionContext) -> None:
        contents = await asyncio.gather(
            *(storage.read_bytes(asset, max_bytes=10_000_000) for asset in self.assets)
        )
        context.logger.info("Read assets", count=len(contents))

The Tilebox Storage client can be reused across task executions. Follow the documented lifetime of other async clients because some clients are tied to the event loop where they were created.

The task runner APIs remain synchronous, and making execute asynchronous does not cause separate workflow tasks to run concurrently within one runner. It only lets one task perform related I/O concurrently.