This article is optional. Python ships asyncio in its standard library, and
if your project uses it, you can skip to the next item. Read it when you meet
anyio or trio in a codebase. You may meet AnyIO soon, because Starlette,
FastAPI and HTTPX use it internally.
Here is a task group in AnyIO:
import anyio
async def fetch(name: str) -> None:
await anyio.sleep(1) # stands in for a network call
print("fetched", name)
async def main() -> None:
async with anyio.create_task_group() as tg:
tg.start_soon(fetch, "users")
tg.start_soon(fetch, "orders")
anyio.run(main) # runs on asyncio, or pass backend="trio" to run on Trio
It works like asyncio.TaskGroup. The block waits for both tasks, and if one
fails, the group cancels the other. One difference is that start_soon takes
the function and its arguments and returns nothing, so you collect results
yourself, for example in a dict.
Three libraries, two event loops
| Library | What it is | Pick it when |
|---|---|---|
| asyncio | The standard library's event loop and tools | You start a new project, or your libraries use asyncio |
| Trio | A separate async library with its own event loop | You want its design, and the libraries you need support it |
| AnyIO | One API that runs on top of asyncio or Trio | You write a library that must work with both, or you prefer its API |
Trio calls its task group a nursery, and asyncio's TaskGroup follows the
same idea. AnyIO code runs on Trio only if every library it calls supports
Trio too. An asyncio-only database driver works on AnyIO's asyncio backend
and fails on Trio.
A cancelled scope stays cancelled
The biggest difference is cancellation. Trio and AnyIO use cancel scopes.
A cancel scope is a with block that you can cancel as a whole, and every
task group has one, tg.cancel_scope. Once a scope is cancelled, every
await inside it raises the cancellation exception, until the code leaves
the scope.
In asyncio, task.cancel() raises CancelledError once, at the task's next
await. If the task catches it, its next await runs normally.
That difference changes how you write cleanup. In a cancelled AnyIO scope,
an await in a finally block is cancelled straight away. To let the
cleanup finish, shield it:
async def worker() -> None:
try:
await anyio.sleep_forever()
finally:
with anyio.move_on_after(1, shield=True):
await close_connection()
shield=True protects the block from the outer cancellation. The
move_on_after(1) timeout still applies, so the cleanup gets at most one
second.
Check your understanding Your app uses asyncio, and you add HTTPX, which uses AnyIO inside. Do you need to rewrite your app with AnyIO? Show answer Hide answer
No. On the asyncio backend, AnyIO runs on the asyncio event loop you already
have. Your asyncio.TaskGroup code and the library's AnyIO code share that
one loop. You need AnyIO in your own code only if you want its API or want to
run on Trio.
Going further
To move a project to Trio, or to support both AnyIO backends, first check
that every library you depend on supports that choice. Then run your tests on
each backend you support. Include tests for failures, cancellation and
shutdown.
Install anyio[trio] to get the Trio backend.
Read more in AnyIO basics, AnyIO cancellation and the Trio reference.