AnyIO

Warnings

• breaking In AnyIO 4, task groups always raise ExceptionGroup (PEP 654) when any child task raises. The old anyio.ExceptionGroup class was removed. Code catching exceptions from task groups must use 'except*' syntax (Python 3.11+) or the 'exceptiongroup' backport on older Pythons.
  Fix: Use 'except* ValueError as eg:' on Python 3.11+, or install the 'exceptiongroup' package and import ExceptionGroup from it on Python 3.10.
• breaking TaskGroup.spawn() was removed in AnyIO 4. Calling it raises AttributeError.
  Fix: Replace 'await tg.spawn(fn, arg)' with the synchronous 'tg.start_soon(fn, arg)' (no await).
• breaking start_blocking_portal() was moved to anyio.from_thread and must be used as a synchronous context manager. Importing it from the top-level anyio namespace raises ImportError.
  Fix: Use: from anyio.from_thread import start_blocking_portal with start_blocking_portal() as portal: portal.call(fn)
• deprecated The 'item_type' keyword argument to create_memory_object_stream() is deprecated and will be removed in a future major version.
  Fix: Use the generic syntax: create_memory_object_stream[MyType](max_buffer_size=N)
• deprecated The 'cancellable' parameter of to_thread.run_sync() is a deprecated alias for 'abandon_on_cancel'. Passing it emits a DeprecationWarning.
  Fix: Replace cancellable=True with abandon_on_cancel=True.
• gotcha anyio.Path cannot be substituted for pathlib.Path. APIs expecting a pathlib.Path will reject an anyio.Path instance; convert with str(path) or pass the underlying pathlib.Path via path._path (private, unstable).
  Fix: Pass str(anyio_path) to third-party APIs expecting pathlib.Path, or open the file with await anyio_path.open() and hand the file object over.
• gotcha Mixing native asyncio tasks (asyncio.create_task) with AnyIO cancel scopes can cause tasks spawned outside AnyIO to escape cancellation, and await inside a finally block can receive multiple CancelledError instances under scoped cancellation.
  Fix: Spawn all concurrent work through anyio.create_task_group().start_soon() and avoid bare asyncio.create_task() inside AnyIO-managed code.

Install

• pip install anyio asyncio backend (default)
• pip install anyio[trio] with Trio backend support
• pip install anyio[trio] pytest-anyio with Trio + built-in pytest plugin

Imports

• run

  import anyio
  anyio.run(main)

  Use anyio.run() to stay backend-agnostic; pass backend='trio' or backend='asyncio' explicitly when needed
• create_task_group

  from anyio import create_task_group
  async with create_task_group() as tg:
      tg.start_soon(fn)

  TaskGroup.spawn() was removed in AnyIO 4; use the synchronous tg.start_soon() instead
• create_memory_object_stream

  from anyio import create_memory_object_stream
  send, recv = create_memory_object_stream[int](max_buffer_size=10)

  The item_type keyword argument is deprecated; use the generic bracket syntax create_memory_object_stream[T]()
• start_blocking_portal

  from anyio.from_thread import start_blocking_portal
  with start_blocking_portal() as portal:
      portal.call(async_fn)

  start_blocking_portal lives in anyio.from_thread, not the top-level anyio namespace; must be used as a context manager
• to_thread.run_sync

  from anyio import to_thread
  await to_thread.run_sync(blocking_fn, abandon_on_cancel=True)

  The 'cancellable' parameter is a deprecated alias for 'abandon_on_cancel'; use abandon_on_cancel instead
• CancelScope

  from anyio import CancelScope
  with CancelScope(deadline=...) as scope:
      ...

  CancelScope is a plain synchronous context manager since AnyIO 3; do NOT use 'async with'
• Path

  from anyio import Path
  path = Path('/tmp/file')
  content = await path.read_text()

  anyio.Path is NOT a drop-in replacement for pathlib.Path; it cannot be passed to APIs expecting pathlib.Path
• get_cancelled_exc_class

  from anyio import get_cancelled_exc_class
  try:
      await something()
  except get_cancelled_exc_class():
      raise

  The cancellation exception differs between backends; use get_cancelled_exc_class() to catch it portably

Quickstart

Demonstrates concurrent task groups, move_on_after timeout scopes, and offloading blocking calls to a worker thread — all backend-agnostic.

import anyio

async def fetch(name: str, delay: float) -> None:
    print(f'{name}: starting')
    await anyio.sleep(delay)
    print(f'{name}: done after {delay}s')

async def main() -> None:
    # Task group: all tasks run concurrently; exceptions surface as ExceptionGroup
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch, 'task-A', 1.0)
        tg.start_soon(fetch, 'task-B', 0.5)

    # Timeout via CancelScope
    with anyio.move_on_after(2.0) as scope:
        await anyio.sleep(5)
    if scope.cancelled_caught:
        print('timed out (moved on)')

    # Run a blocking call in a thread
    import time
    result = await anyio.to_thread.run_sync(time.strftime, '%X')
    print('wall-clock time from thread:', result)

if __name__ == '__main__':
    # Default backend is asyncio; swap to 'trio' with no code changes
    anyio.run(main, backend='asyncio')