To https://github.com/graingert/efibootnext
Boot Windows from a GRUB menu entry without breaking BitLocker TPM measurements.
To https://github.com/graingert/efibootnext
Boot Windows from a GRUB menu entry without breaking BitLocker TPM measurements.
| from collections.abc import AsyncGenerator | |
| from contextlib import asynccontextmanager | |
| from fastapi import FastAPI, Request | |
| from pydantic import BaseModel | |
| from typing import TypedDict | |
| import anyio | |
| import anyio.abc | |
| class State(TypedDict): |
| @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 |
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.
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:
| import sys | |
| from typing import Self, cast | |
| import dataclasses | |
| @dataclasses.dataclass(slots=False) | |
| class Node: | |
| _parent: Self | None | |
| MessagePump = Node | |
| DOMNode = Node |
| import sys | |
| from typing import Self, cast | |
| import dataclasses | |
| @dataclasses.dataclass(slots=False) | |
| class Node: | |
| _parent: Self | None | |
| MessagePump = Node | |
| DOMNode = Node |
| import sys | |
| from typing import Self, cast | |
| import dataclasses | |
| @dataclasses.dataclass(slots=False) | |
| class Node: | |
| _parent: Self | None | |
| MessagePump = Node | |
| DOMNode = Node |
| """Middleware.""" | |
| from __future__ import annotations | |
| import logging | |
| import pathlib | |
| from django.conf import settings | |
| from whitenoise.middleware import WhiteNoiseMiddleware # type: ignore[import-untyped] |
| 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: |