AnyIO

AnyIO lets you write asynchronous code in Python that works with different async backends.

Introduction
Basic Usage
Task Group
Pending

Introduction

AnyIO is an asynchronous networking and concurrency library that works on top of either asyncio or Trio. It implements Trio-like structured concurrency (SC) on top of asyncio and works in harmony with the native SC of Trio itself.

uv init anyio-basic
cd anyio-basic
uv add anyio

Basic Usage

Let’s create a simple example.

Make a file called task.py and add this code:

import anyio

async def do_task():
    print("Start ... ", end="")
    await anyio.sleep(2)
    print("end!")


async def main():
    await do_task()

if __name__ == "__main__":
    anyio.run(main)

When working with async code you normally use async and await keywords.

The main rules are that:

You can only use await inside async functions.
To call an async function, you need to use await in front of it.

In this case, we can use await inside of main() because it is an async function.

Call Async Functions

The function do_task() also needs to be declared with async def for us to be able to await for its result when calling it.

do_task() could be talking to a database, calling an API, or something else that needs to wait for something.

For this example, we simulate that by making do_task() wait there for 2 seconds:

async def do_task():
    print("Start ... ", end="")
    await anyio.sleep(2)
    print("end!")

Run the main function

As main() is an async function, we can’t call it directly because we can’t await it.

Instead, we call it with anyio.run():

if __name__ == "__main__":
    anyio.run(main)

anyio.run() will do everything necessary to call main(), handling all the await parts, and waiting there until it finishes.

By default, AnyIO uses asyncio, but you can switch to trio without changing your code.

Summary

This example highlights the basic structure of AnyIO programs:

define functions with async def
use await to pause operations
start the program with anyio.run().

Task Group

Asynchronous programming allows you to run multiple tasks concurrently, improving performance by handling operations simultaneously, especially for I/O-bound tasks like file reading, network requests, or database queries.

Task groups let you manage and execute tasks concurrently, ensuring all tasks finish before moving forward.

import anyio
import time

async def task(name, delay):
    print(f"Task {name} starting")
    await anyio.sleep(delay)
    print(f"Task {name} completed after {delay} seconds")
    return name, delay

async def main():
    start_time = time.time()

    async with anyio.create_task_group() as tg:
        # Start multiple tasks that run at the same time
        tg.start_soon(task, "A", 2)
        tg.start_soon(task, "B", 1)
        tg.start_soon(task, "C", 3)

    end_time = time.time()
    print(f"All tasks completed in {end_time - start_time:.2f} seconds")

if __name__ == "__main__":
    anyio.run(main)

The create_task_group() method creates a group of tasks that run together.

You add tasks to the group with start_soon().

When you exit the task group, it waits for all tasks to finish.

Run this script:

uv run src/task.py

You’ll see something like:

Python - AnyIO