AnyIO lets you write asynchronous code in Python that works with different async backends.
It gives you a single interface to work with asyncio, trio, and other async libraries. This means you can write your code once and run it anywhere.
This guide will show you how to use AnyIO effectively in your Python projects.
\Prerequisites
Ensure you have installed Python 3.7 or a newer version on your computer. You should already know basic Python and have some experience with asynchronous programming.
Step 1 — Getting started with AnyIO
Start by creating a new folder for your project and setting up a Python virtual environment:
mkdir anyio-tutorial && cd anyio-tutorial
python3 -m venv venv
Activate your virtual environment:
source venv/bin/activate
Install AnyIO:
pip install anyio
Let's create a simple example. Make a file called main.py and add this code:
import anyio
async def hello_world():
print("Hello, world!")
await anyio.sleep(1)
print("Goodbye, world!")
async def main():
await hello_world()
if __name__ == "__main__":
anyio.run(main)
This example highlights the basic structure of AnyIO programs: define functions with async def, use await to pause operations, and start the program with anyio.run(). By default, AnyIO uses asyncio, but you can easily switch to trio without changing your code.
Run it with:
python main.py
You'll see:
Hello, world!
Goodbye, world!
Now that you've seen how to set up a basic AnyIO program and understand its structure, let's move on to the next step.
Step 2 — Working with tasks and concurrency
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.
In this step, you'll learn to use AnyIO’s task groups to run multiple tasks simultaneously. Task groups let you manage and execute tasks concurrently, ensuring all tasks finish before moving forward.
Let's see how to implement this using AnyIO:
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() 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:
python main.py
You'll see something like:
Task A starting
Task B starting
Task C starting
Task B completed after 1 seconds
Task A completed after 2 seconds
Task C completed after 3 seconds
All tasks completed in 3.00 seconds
Notice how the tasks run at the same time, not one after another. If they ran one after another, it would take 6 seconds (2+1+3). But running them together takes only 3 seconds - just the time needed for the longest task.