repository·master·Indexed 25 days ago
https://github.com/agronholm/anyioA high-level concurrency and networking framework for Python that provides a unified API for both asyncio and Trio. AnyIO emphasizes structured concurrency through Task Groups and a level cancellation model using cancel scopes. It includes utilities for running synchronous code in worker threads or processes, asynchronous file I/O, stream-based data transfer, and various synchronization primitives.
AnyIO is an asynchronous networking and concurrency library designed to work on top of either asyncio or Trio. It provides a unified API that allows applications and libraries to run unmodified on either backend.
A key feature is its implementation of structured concurrency (SC) on top of asyncio, which mimics the native SC behavior found in Trio. This allows for incremental adoption into existing codebases without requiring a full refactoring.
asyncio. While asyncio is the standard library implementation, AnyIO addresses several design issues and missing features, particularly regarding structured concurrency and task management. Even for application development (not just library development), AnyIO offers more robust primitives for managing task lifecycles and synchronization.AnyIO supports TLS (Transport Layer Security) to provide authenticity and confidentiality for TCP streams. TLS is typically established immediately after a connection is made via a handshake that involves certificate exchange and hostname verification.
To implement TLS, you can use TLSListener to wrap a standard TCP listener on the server side, or use the ssl_context argument when connecting via connect_tcp on the client side.
import ssl
from anyio import create_tcp_listener, run
from anyio.streams.tls import TLSListener
async def handle(client):
async with client:
name = await client.receive()
await client.send(b'Hello, %s\n' % name)
async def main():
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain(certfile='cert.pem', keyfile='key.pem')
listener = TLSListener(await create_tcp_listener(local_port=1234), context)
await listener.serve(handle)
run(main)__aenter__ and __aexit__ (which requires managing state and handling exceptions in __aenter__), you can use these mix-ins to write logic similar to @contextlib.asynccontextmanager. Note that using these mix-ins sacrifices re-entrancy.By default, when a task is spawned, the context (from contextvars) is copied to the new task.
Important: The context copied is the context of the task that calls TaskGroup.start() or TaskGroup.start_soon(), not the context of the task group's host task.
To run a task in a specific, custom context instead of inheriting the caller's context, pass a contextvars.Context object to the context keyword argument in TaskGroup.create_task().
Connectables. This abstraction allows you to produce connected streams (either object-oriented or bytes-oriented). This is particularly useful for network clients, as it allows you to abstract the connection mechanism, making it easier to customize or mock connections without using monkey patching.Cancel scopes must be entered and exited in LIFO (Last-In, First-Out) order. Violating this can lead to stack corruption and unpredictable behavior.
Risky patterns to avoid:
__enter__ and __exit__ in the wrong order.[Async]ExitStack in a way that doesn't mirror nested context managers.Safe pattern for async context managers:
It is generally safe to use task groups or cancel scopes within an @asynccontextmanager as long as the host task remains running throughout the entire lifecycle of the context manager.
from contextlib import asynccontextmanager
from anyio import create_task_group
# Okay in most cases!
@asynccontextmanager
async def some_context_manager():
async with create_task_group() as tg:
tg.start_soon(foo)
yieldAnyIO provides a wide range of asynchronous primitives and networking capabilities:
Happy eyeballs algorithm for more robust TCP connections.async/await style UDP sockets (avoiding the Transport/Protocol pattern required by asyncio).functools and itertools modules.A stapled stream combines a compatible receive stream and send stream into a single bidirectional stream object.
There are two variants:
StapledByteStream: Combines a ByteReceiveStream and a ByteSendStream.StapledObjectStream: Combines an ObjectReceiveStream and a compatible ObjectSendStream.In asyncio, if a task is scheduled to resume with a value (the await is about to yield a result) and is cancelled at that exact moment, the CancelledError is raised instead of yielding the result, causing the data to be lost.
AnyIO's cancel scopes ensure that a task which is scheduled to resume will be able to process the result of the await before the cancellation is applied.
import asyncio
import anyio
async def receive(f):
print(await f)
await asyncio.sleep(1)
print("The task will be cancelled before this is printed")
async def main():
f = asyncio.get_running_loop().create_future()
async with anyio.create_task_group() as tg:
tg.start_soon(receive, f)
await asyncio.sleep(0) # make sure the task has started
f.set_result("hello")
tg.cancel_scope.cancel()
# Output: "hello"
asyncio.run(main())EndOfStream exception instead of returning an empty bytes object.While asyncio.TaskGroup (introduced in Python 3.11) provides basic structured concurrency, it has several limitations compared to AnyIO's task groups:
asyncio.TaskGroup does not provide a way to cancel or list all contained tasks, forcing developers to manually track tasks. AnyIO task groups include a cancel scope that can cancel all child tasks regardless of where they were launched.asyncio.TaskGroup lacks a built-in mechanism to wait until a newly launched task signals it is ready. AnyIO provides patterns for waiting until a child task is initialized and ready to proceed.