AnyIO , a Python library providing structured concurrency primitives on top of asyncio

See also

• https://github.com/agronholm/anyio

• https://github.com/agronholm/anyio/graphs/contributors

• https://anyio.readthedocs.io/en/latest/index.html

• https://anyio.readthedocs.io/en/latest/tasks.html

• https://anyio.readthedocs.io/en/latest/synchronization.html

• https://anyio.readthedocs.io/en/latest/streams.html

• https://anyio.readthedocs.io/en/latest/api.html#timeouts-and-cancellation

• https://anyio.readthedocs.io/en/latest/api.html

Description

AnyIO is a asynchronous compatibility API that allows applications and libraries written against it to run unmodified on asyncio, curio and trio.

It bridges the following functionality:

• Task groups

• Cancellation

• Threads

• Signal handling

• Asynchronous file I/O

• Subprocesses

• Inter-task synchronization and communication (locks, conditions, events, semaphores, object streams)

• High level networking (TCP, UDP and UNIX sockets)

You can even use it together with native libraries from your selected backend in applications.

Doing this in libraries is not advisable however since it limits the usefulness of your library.

AnyIO comes with its own pytest plugin which also supports asynchronous fixtures.

It even works with the popular Hypothesis library .

Articles

Structured concurrency in Python with AnyIO by Matt Westcott (August 17, 2020)

See also

• https://mattwestcott.co.uk/blog/structured-concurrency-in-python-with-anyio

• https://en.wikipedia.org/wiki/Structured_concurrency

• https://github.com/mjwestcott

• https://anyio.readthedocs.io/en/latest/tasks.html

By now you might be familiar with the term structured concurrency .

It’s a way to write concurrent programs that’s easier than manually taking care of the lifespan of concurrent tasks.

The best overview is Notes on structured concurrency by Nathaniel Smith (or his video if you prefer).

This post is about AnyIO, a Python library providing structured concurrency primitives on top of asyncio .

This is important because your project is probably dependent on an asyncio-based library (e.g. Starlette or asyncpg) if you are building a non-trivial production system, so you sadly can’t use Trio.

The current state of async Python

Python has three well-known concurrency libraries built around the async/await syntax: asyncio , Curio, and Trio.

The first, asyncio , was designed by Guido van Rossum and is included in the Python standard library.

Since it’s the default, the overwhelming majority of async applications and libraries are written with asyncio . It has, however, received mixed reviews.

The second and third are attempts to improve on asyncio , by David Beazley and Nathaniel Smith respectively.

Neither of them has an ecosystem to match asyncio , but Trio, in particular, has very effectively demonstrated the benefits of structured concurrency .

Indeed, the Python core team appears to have been persuaded that asyncio should move in that direction since at least as far back as 2017.

But as of today, asyncio’s progress has been slow .

Python’s async ecosystem is therefore in a strange place, as described by Nathaniel on Stack Overflow

AnyIO to the rescue

The AnyIO library by Alex Grönholm describes itself as follows:

an asynchronous compatibility API that allows applications and
libraries written against it to run unmodified on asyncio, curio and trio.

So you might assume it’s only useful if you want to write code that is agnostic between those backends.

I think that’s a pretty niche scenario: it applies if you are, for example, creating a new database driver and you want it to be used as widely as possible.

However, it’s much more likely that you are a consumer of an existing database driver, one that is almost certainly written against asyncio (e.g. asyncpg, aredis, etc.).

If you continue reading the AnyIO docs, you learn that:

You can even use it together with native libraries from your
selected backend in applications.

Doing this in libraries is not advisable however since it limits
the usefulness of your library.

This is the most important part.

You can in fact use it directly with libraries written for asyncio.

In other words: AnyIO is an implementation of structured concurrency for asyncio .

The author doesn’t recommend using it like this if you’re writing a library because it runs counter to the ostensible point of AnyIO: your code will not be backend-agnostic.

I’m sympathetic to his point, since if everyone builds on top of AnyIO, then the whole community benefits from greater flexibility.

But if you’re a pragmatic engineer building production systems, you can ignore that advice. You can use AnyIO as an enhancement to asyncio, when creating libraries as well as applications.

Alex has modelled AnyIO’s API after Trio, so this is a significant improvement.

An example: graceful task shutdown

See also

• https://github.com/mjwestcott/runnel

• https://runnel.dev/architecture.html

I recently released Runnel , a distributed stream processing library for Python built on top of Redis Streams.

It is dependent on aredis, an asyncio Redis driver and I want it to be usable by applications with other asyncio dependencies.

I built the first prototype directly with asyncio, but working with its gather and wait primitives was very frustrating. Neither of them will cancel other tasks when one completes. And building correct cancellation semantics on top is non-trivial to say the least: just look at the AnyIO implementation.

The main problem was cleaning up resources during a graceful program shutdown. In Runnel, workers hold locks implemented in Redis for partitions of an event stream. They should be released when a worker exits, so that others can quickly acquire them and continue to work through the backlog of events. Here’s the most basic clean up example in AnyIO:

import anyio


async def task(n):
    await anyio.sleep(n)


async def main():
    try:
        async with anyio.create_task_group() as tg:
            await tg.spawn(task, 1)
            await tg.spawn(task, 2)
    finally:
        # e.g. release locks
        print('cleanup')


anyio.run(main)

This program will run for 2 seconds before running the finally block. In a more realistic scenario, the tasks might run forever until one of them raises an exception.

Using AnyIO, the remaining tasks will be cancelled and the finally block will run before control flow continues beyond the TaskGroup’s scope. (See here for more information about this API.)

If you’re interested in this architecture, check out the Runnel docs .

For our purposes, what matters is that each box is an independent asyncio task running forever or until a specific event or timeout occurs.

Failures in tasks should be handled by their parent: normally by propagating the exception after cancelling all siblings and cleaning up.

Sometimes restarting the task is appropriate.

Implementing this scheme became much easier when I switched from native asyncio to the AnyIO abstractions. (See, for example, the task groups here 1 , here 2 and here 3 .)

In addition to the task groups described above, AnyIO also provides other primitives to replace the native asyncio ones if you want to benefit from structured concurrency’s cancellation semantics:

• Synchronisation primitives (locks, events, conditions)

• Streams (similar to queues)

• Timeouts (e.g. move_on_after, fail_after)

• … and more

Summary: use AnyIO on top of asyncio for the best of both worlds

See also

• Structured Concurrency

Asyncio is hard to use because it doesn’t support structured concurrency , but has very wide library support.

Trio is easy to use, but it’s not the community default so its library ecosystem cannot compete.

This post presents an alternative: use AnyIO on top of asyncio for the best of both worlds .