graingert

@app.get("/news-and-weather")
@contextlib.asynccontextmanager
async def news_and_weather() -> AsyncGenerator[MemoryObjectRecieveStream[bytes]]:
    async with anyio.create_task_group() as tg:
        tx, rx = anyio.create_memory_object_stream[bytes]()
        with tx, rx:
             tg.start_soon(ws_stream, "ws://example.com/news", tx.clone())
             tg.start_soon(ws_stream, "ws://example.com/weather", tx.clone())
             tx.close()
             yield rx

graingert

Why You Should Use AnyIO (And Why You Might Already Have It)

Abstract

Async Python is fragmented—but you probably already have the solution installed. AnyIO is a portability layer for asyncio and Trio that fixes critical cancellation bugs and provides structured concurrency. If you use httpx, FastAPI, or Jupyter, AnyIO is already in your environment powering your HTTP clients, web frameworks, and notebooks.

This talk reveals AnyIO's level-triggered cancellation (fixing asyncio's dangerous edge-triggered behavior that causes silent hangs), demonstrates structured concurrency patterns with task groups, and shows practical tools like memory object streams for producer-consumer workflows. You'll learn why major libraries chose AnyIO—and how to use it directly for more reliable concurrent code.

Target audience: Intermediate Python developers working with async code who want portable, maintainable concurrent programs.

graingert

Elaboration on the Key Difference

The Fundamental Distinction

The core difference is about who holds responsibility for the reference during a function call:

1. Python-to-Python Calls (Reference Move)

When Python code calls a Python function through the bytecode interpreter:

• The caller's evaluation stack transfers ownership of the argument references to the callee

graingert

import sys
from typing import Self, cast
import dataclasses

@dataclasses.dataclass(slots=False)
class Node:
    _parent: Self | None

MessagePump = Node
DOMNode = Node

graingert

import sys
from typing import Self, cast
import dataclasses

@dataclasses.dataclass(slots=False)
class Node:
    _parent: Self | None

MessagePump = Node
DOMNode = Node

graingert

import sys
from typing import Self, cast
import dataclasses

@dataclasses.dataclass(slots=False)
class Node:
    _parent: Self | None

MessagePump = Node
DOMNode = Node

graingert

"""Middleware."""

from __future__ import annotations

import logging
import pathlib

from django.conf import settings
from whitenoise.middleware import WhiteNoiseMiddleware  # type: ignore[import-untyped]

graingert

import socket
import asyncio
import selectors


async def wait_readable(s):
    event = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_reader(s, event.set)
    try:

graingert

import anyio
import socket


async def demo():
    s1, s2 = socket.socketpair()
    with s1, s2:
        s1.setblocking(False)
        s2.setblocking(False)
        async with anyio.create_task_group() as tg:

graingert

import anyio.streams.buffered
import anyio.abc
import socket
import functools


async def pipe(reader, writer):
    raw_socket = reader.extra_attributes[anyio.abc.SocketAttribute.raw_socket]()
    try:
        while True: