Skip to content

Document async tasks #131

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lukasbindreiter merged 1 commit into from
Aug 12, 2026
Merged

Document async tasks #131

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions api-reference/python/tilebox.workflows/Task.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ icon: code

```python
class Task:
def execute(context: ExecutionContext) -> None
def execute(context: ExecutionContext) -> Awaitable[None] | None

@staticmethod
def identifier() -> tuple[str, str]
Expand All @@ -17,10 +17,10 @@ Inheriting also automatically applies the dataclass decorator.
## Methods

```python
def Task.execute(context: ExecutionContext) -> None
def Task.execute(context: ExecutionContext) -> Awaitable[None] | None
```

The entry point for the execution of the task.
The entry point for the execution of the task. Define it as either `def execute(...) -> None` or `async def execute(...) -> None`. The runner waits for an asynchronous method to complete.

```python
@staticmethod
Expand Down
139 changes: 27 additions & 112 deletions sdks/python/async.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,130 +97,45 @@ contents = await storage.read_bytes(assets["thumbnail"], max_bytes=10_000_000)

See [Read and download assets](/datasets/assets-and-storage/read-and-download) for streaming, downloads, and GeoTIFF window reads.

## Fetching data concurrently
## Downloading assets concurrently

The primary benefit of the async client is that it allows concurrent requests, enhancing performance.
In below example, data is fetched from multiple collections. The synchronous approach retrieves data sequentially, while the async approach does so concurrently, resulting in faster execution.

<CodeGroup>
```python Python (Sync)
# Example: fetching data sequentially

# switch to the async example to compare the differences
import time
from tilebox.datasets import Client
from tilebox.datasets.sync.timeseries import TimeseriesCollection

client = Client()
datasets = client.datasets()
collections = datasets.open_data.copernicus.landsat8_oli_tirs.collections()

def stats_for_2020(collection: TimeseriesCollection) -> None:
"""Fetch data for 2020 and print the number of data points that were loaded."""
data = collection.query(temporal_extent=("2020-01-01", "2021-01-01"), show_progress=True)
n = data.sizes['time'] if 'time' in data else 0
return (collection.name, n)

start = time.monotonic()
results = [stats_for_2020(collections[name]) for name in collections]
duration = time.monotonic() - start

for collection_name, n in results:
print(f"There are {n} datapoints in {collection_name} for 2020.")
print(f"Fetching data took {duration:.2f} seconds")
```

```python Python (Async)
# Example: fetching data 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 Python
import asyncio
import time
from tilebox.datasets.aio import Client
from tilebox.datasets.aio.timeseries import TimeseriesCollection

client = Client()
datasets = await client.datasets()
collections = await datasets.open_data.copernicus.landsat8_oli_tirs.collections()
await asyncio.gather(
storage.download(assets["red"], "red.tif"),
storage.download(assets["green"], "green.tif"),
storage.download(assets["blue"], "blue.tif"),
)
```

async def stats_for_2020(collection: TimeseriesCollection) -> None:
"""Fetch data for 2020 and print the number of data points that were loaded."""
data = await collection.query(temporal_extent=("2020-01-01", "2021-01-01"), show_progress=True)
n = data.sizes['time'] if 'time' in data else 0
return (collection.name, n)
## Async workflows

start = time.monotonic()
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()`.

# Initiate all requests concurrently
requests = [stats_for_2020(collections[name]) for name in collections]
# Wait for all requests to finish in parallel
results = await asyncio.gather(*requests)
For example, a task can read a small group of assets concurrently:

duration = time.monotonic() - start
```python Python
import asyncio

for collection_name, n in results:
print(f"There are {n} datapoints in {collection_name} for 2020.")
print(f"Fetching data took {duration:.2f} seconds")
```
</CodeGroup>
from tilebox.datasets.assets import Asset
from tilebox.storage.aio import Client as StorageClient
from tilebox.workflows import ExecutionContext, Task

The output demonstrates that the async approach runs approximately 30% faster for this example. With `show_progress` enabled, the progress bars update concurrently.
storage = StorageClient()

<CodeGroup>
```plaintext Python (Sync)
There are 19624 datapoints in L1GT for 2020.
There are 1281 datapoints in L1T for 2020.
There are 65313 datapoints in L1TP for 2020.
There are 25375 datapoints in L2SP for 2020.
Fetching data took 10.92 seconds
```
class ReadAssets(Task):
assets: list[Asset]

```plaintext Python (Async)
There are 19624 datapoints in L1GT for 2020.
There are 1281 datapoints in L1T for 2020.
There are 65313 datapoints in L1TP for 2020.
There are 25375 datapoints in L2SP for 2020.
Fetching data took 7.45 seconds
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))
```
</CodeGroup>

## Async workflows

The Tilebox workflows Python client does not have an async client. This is because workflows are designed for distributed and concurrent execution outside a single async event loop. But within a single task, you may use still use`async` code to take advantage of asynchronous execution, such as parallel data loading. You can achieve this by wrapping your async code in `asyncio.run`.

Below is an example of using async code within a workflow task.

<CodeGroup>
```python Python (Async)
import asyncio
import xarray as xr

from tilebox.datasets.aio import Client as DatasetsClient
from tilebox.datasets.query import TimeIntervalLike
from tilebox.workflows import Task, ExecutionContext

class FetchData(Task):
def execute(self, context: ExecutionContext) -> None:
# The task execution itself is synchronous
# But we can leverage async code within the task using asyncio.run

# This will fetch three months of data in parallel
data_jan, data_feb, data_mar = asyncio.run(load_first_three_months())

async def load_data(interval: TimeIntervalLike):
datasets = await DatasetsClient().datasets()
collections = await datasets.open_data.copernicus.landsat8_oli_tirs.collections()
return await collections["L1T"].query(temporal_extent=interval)

async def load_first_three_months() -> tuple[xr.Dataset, xr.Dataset, xr.Dataset]:
jan = load_data(("2020-01-01", "2020-02-01"))
feb = load_data(("2020-02-01", "2020-03-01"))
mar = load_data(("2020-03-01", "2020-04-01"))
# load the three months in parallel
jan, feb, mar = await asyncio.gather(jan, feb, mar)
return jan, feb, mar
```
</CodeGroup>
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.

<Tip>
If you encounter an error like `RuntimeError: asyncio.run() cannot be called from a running event loop`, it means you're trying to start another asyncio event loop (with `asyncio.run`) from within an existing one. This often happens in Jupyter notebooks since they automatically start an event loop. A way to resolve this is by using [nest-asyncio](https://pypi.org/project/nest-asyncio/).
</Tip>
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.
4 changes: 4 additions & 0 deletions workflows/concepts/tasks.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,10 @@ For python, the key components of this task are:
</Accordion>
</AccordionGroup>

<Tip>
Python tasks can also define `execute` with `async def`. This lets you await asynchronous APIs directly inside a task without calling `asyncio.run()`. See [Async support](/sdks/python/async#async-workflows) for an example.
</Tip>

For Go, the key components are:

<AccordionGroup>
Expand Down