# Async support

## Why use async?

When working with external datasets, such as [Tilebox datasets](/docs/datasets/concepts/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.

**Python (Sync)**

```python title="Python (Sync)"
from tilebox.datasets import Client

# This client is synchronous
client = Client()
```

**Python (Async)**

```python title="Python (Async)"
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.

**Python (Sync)**

```python title="Python (Sync)"
# 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)
```

**Python (Async)**

```python title="Python (Async)"
# 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)
```

Jupyter notebooks and similar interactive environments support asynchronous code execution. You can use
`await some_async_call()` as the output of a code cell.

## Accessing assets asynchronously

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

```python title="Python"
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](/docs/datasets/assets-and-storage/read-and-download) 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:

```python title="Python"
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:

```python title="Python"
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.