Python AnyIO Portability — Deep Dive

How AnyIO Detects the Backend

AnyIO uses a simple but clever mechanism: it checks what’s currently running. When you call any AnyIO function, it inspects the running event loop:

# Simplified version of AnyIO's backend detection
def get_current_backend():
    try:
        import trio
        trio.lowlevel.current_trio_token()
        return "trio"
    except (ImportError, RuntimeError):
        pass
    
    import asyncio
    asyncio.get_running_loop()  # Raises if no loop
    return "asyncio"

The real implementation uses sniffio, a tiny library that provides sniffio.current_async_library(). This is the standard way to detect which async framework is running.

Structured Concurrency on asyncio

One of AnyIO’s most valuable features is bringing Trio-style structured concurrency to asyncio. In pure asyncio (before 3.11), tasks could escape their parent’s scope:

# Pure asyncio — task outlives the function that created it
async def leaky():
    asyncio.create_task(background_work())  # Who owns this task?
    return "done"  # Function returns, but background_work keeps running

AnyIO’s task groups prevent this:

# AnyIO — all tasks are bounded by the group
async def contained():
    async with anyio.create_task_group() as tg:
        tg.start_soon(background_work)
    # background_work is guaranteed to be done here
    return "done"

Python 3.11’s asyncio.TaskGroup was directly inspired by this pattern.

Porting asyncio Code to AnyIO

Common Translations

asyncio	AnyIO
asyncio.sleep(n)	anyio.sleep(n)
asyncio.gather(*coros)	Task group with tg.start_soon()
asyncio.create_task(coro)	tg.start_soon(coro) (inside task group)
asyncio.wait_for(coro, timeout)	with anyio.fail_after(timeout): await coro
asyncio.Event()	anyio.Event()
asyncio.Lock()	anyio.Lock()
asyncio.Semaphore(n)	anyio.Semaphore(n)
asyncio.Queue()	anyio.create_memory_object_stream()
loop.run_in_executor(None, fn)	await anyio.to_thread.run_sync(fn)

The Gather Pattern

asyncio’s gather has no direct equivalent. AnyIO uses task groups with shared state:

# asyncio style
results = await asyncio.gather(fetch_a(), fetch_b(), fetch_c())

# AnyIO style
results = {}
async with anyio.create_task_group() as tg:
    async def run(key, coro):
        results[key] = await coro
    
    tg.start_soon(run, "a", fetch_a())
    tg.start_soon(run, "b", fetch_b())
    tg.start_soon(run, "c", fetch_c())
# results is now {"a": ..., "b": ..., "c": ...}

Thread Integration

# Run blocking code in a thread
result = await anyio.to_thread.run_sync(blocking_function, arg1, arg2)

# Run async code from a sync thread
anyio.from_thread.run(async_function, arg1, arg2)

# Run sync callback from async context with cancellation support
await anyio.to_thread.run_sync(blocking_io, cancellable=True)

Building a Portable Library

Here’s a real-world pattern for a library that works on both backends:

# mylib/client.py
import anyio

class AsyncClient:
    def __init__(self, base_url: str, timeout: float = 30.0):
        self.base_url = base_url
        self.timeout = timeout
    
    async def __aenter__(self):
        self._stream = await anyio.connect_tcp(
            self.base_url, 443, tls=True
        )
        return self
    
    async def __aexit__(self, *exc):
        await self._stream.aclose()
    
    async def request(self, path: str) -> bytes:
        with anyio.fail_after(self.timeout):
            await self._stream.send(
                f"GET {path} HTTP/1.1\r\nHost: {self.base_url}\r\n\r\n".encode()
            )
            return await self._stream.receive(65536)

# Works on both:
# anyio.run(main, backend="asyncio")
# anyio.run(main, backend="trio")

Testing Portable Libraries

Use pytest-anyio to test against both backends:

# conftest.py
import pytest

@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    return request.param

# test_client.py
import pytest
import anyio

@pytest.mark.anyio
async def test_client_connect():
    async with AsyncClient("example.com") as client:
        response = await client.request("/")
        assert len(response) > 0

This runs every test twice — once on asyncio, once on Trio.

Advanced: Cancel Scopes in Depth

AnyIO’s cancel scopes are more powerful than simple timeouts:

async def complex_cancellation():
    # Nested scopes
    with anyio.fail_after(30) as outer_scope:
        with anyio.fail_after(5) as inner_scope:
            await fast_operation()
        
        # inner_scope expired, but outer is still active
        with anyio.move_on_after(10) as optional_scope:
            await optional_enrichment()
        
        if optional_scope.cancelled_caught:
            # Enrichment timed out, use defaults
            pass

Shielding from Cancellation

Sometimes you need an operation to complete even if the parent scope is cancelled:

async def critical_cleanup():
    with anyio.CancelScope(shield=True):
        # This runs even if the outer scope is cancelled
        await save_state_to_disk()
        await notify_peers()

File I/O

AnyIO provides async file operations that work on both backends:

async def process_file(path):
    async with await anyio.open_file(path) as f:
        contents = await f.read()
    return contents

# Directory operations
async def list_dir(path):
    return await anyio.Path(path).iterdir()

Under the hood, these use thread pools — file I/O is inherently blocking, but wrapping it in threads preserves the async programming model.

Performance Comparison

In benchmarks measuring HTTP request throughput:

• Pure asyncio: Baseline (100%)
• AnyIO on asyncio: ~98-99% of pure asyncio (1-2% overhead)
• AnyIO on Trio: ~95-97% of pure asyncio (Trio’s safety checks add small overhead)

The difference is negligible for I/O-bound workloads, where network latency dominates. CPU-bound async code (which is an antipattern anyway) would show larger gaps.

Interoperability Escape Hatches

When you need backend-specific features:

import anyio
import sniffio

async def with_backend_specific():
    backend = sniffio.current_async_library()
    
    if backend == "asyncio":
        import asyncio
        loop = asyncio.get_running_loop()
        # Use asyncio-specific features
    elif backend == "trio":
        import trio
        # Use Trio-specific features

This should be rare — if you find yourself doing this often, you might not need AnyIO.

When AnyIO Shines vs When to Skip It

Shines:

• Library code used by many projects (httpx, Starlette, encode/databases)
• Teams transitioning from asyncio to Trio (or evaluating both)
• Projects wanting structured concurrency without Python 3.11’s TaskGroup

Skip:

• Application code with a firm backend choice
• asyncio-only ecosystem (Django, most AWS tooling)
• Performance-critical inner loops where even 1-2% matters

One thing to remember: AnyIO’s real power is enabling the async library ecosystem to be backend-agnostic. It brings structured concurrency to asyncio, provides clean abstractions for networking and synchronization, and imposes near-zero overhead. If you build async libraries, AnyIO is how you avoid forcing your users to pick a specific backend.