Servers, async for, Curio

Source: Fluent Python, 2nd Edition — Chapter 22 by Luciano Ramalho (mojifinder servers through “how async doesn’t”)

22a ended at client awaits. This half: ASGI/FastAPI and a pure asyncio TCP toy, then async for, async generators, @asynccontextmanager, PEP 530 comprehensions, Curio, and why “I/O-bound system” is a myth. No live uvicorn/Telnet in the sandbox.

Inverted index (mojifinder)

Search Unicode by words in unicodedata.name. Documents = characters. InvertedIndex: defaultdict from each name-word → set of chars. DIGIT → hundreds of glyphs; SEVEN includes clocks and dingbats. search splits the query and intersects the posting lists: "cat" 14, "face" 171, "cat face" 10. Classic IR. Concurrent servers only call this; the index itself is sync.

Starter returns '' so both tests fail until you intersect.

FastAPI (web_mojifinder.py)

ASGI (Chapter 20b). SPA: first HTML, then JS hits JSON. FastAPI (Starlette → asyncio): decorators, type hints, OpenAPI /docs. Run: uvicorn web_mojifinder:app --reload. No main — the server loads app.

app = FastAPI(title=..., description=...). CharName(BaseModel) with char / name. init stashes InvertedIndex and form.html on app.state (pathlib /). @app.get('/search', response_model=list[CharName]) async def search(q: str): — params not in the path come from the query string; missing q → 422. Return a generator of dicts; pydantic serializes. @app.get('/', response_class=HTMLResponse, include_in_schema=False) def form() can be sync. search has no await here — still async def so a real app can await a DB. response_model= not a return annotation: the function may return dicts/ORM rows; FastAPI still limits fields.

Serving HTML from the app is a demo hack. Prod: Traefik/CDN/static on the proxy. Project generators assume that.

asyncio TCP (tcp_mojifinder.py)

Telnet/netcat, bytes, no HTTP. main builds the index (~0.6 s), asyncio.run(supervisor), KeyboardInterrupt → clean message.

server = await asyncio.start_server(
    functools.partial(finder, index),  # callback must be (reader, writer) only
    host, port)
await server.serve_forever()  # without this, supervisor returns and the process exits

start_server already accepts connections; serve_forever keeps the supervisor parked. Each client gets a new finder task — many sessions, one loop.

finder: peername; loop: writer.write(PROMPT) (not a coro — buffers), await writer.drain() (is a coro — real I/O), data = await reader.readline(), decode UTF-8 (CTRL-C junk → '\x00'), break if first codepoint < 32, await search(...), then writer.close(); await writer.wait_closed().

search: index.search, writelines of encoded rows (not a coro), await drain(), status line, drain again. Mix of plain vs awaitable on StreamWriter is the gotcha; docs now label coroutines. Lower layer: Transports/Protocols (Twisted-shaped).

Quiz

write vs drain

1 / 2

StreamWriter.write vs drain

Pick an answer to see if you are right.

async for

Like async with vs __aenter__: async for needs an asynchronous iterable — __aiter__ is a regular method returning an asynchronous iterator. That iterator: __anext__ coroutine (awaitable) and usually __aiter__ → self. Don’t make the iterable its own iterator (Chapter 17).

aiopg cursor: async for row in cur so a fat SELECT can yield the loop between row batches.

Async generator functions

async def + yield → async generator (parallel to sync generators vs iterator classes). for over as_completed is sync — that API is a classic generator. Then yield await coro.

async def multi_probe(domains) -> AsyncIterator[Result]:
    coros = [probe(d, loop) for d in domains]
    for coro in asyncio.as_completed(coros):
        yield await coro   # parsed as yield (await coro)

probe may take an optional loop to avoid repeating get_running_loop. Return NamedTuple. domaincheck.py: async for domain, found in multi_probe(...): two columns.

python -m asyncio (3.8+): loop already running; top-level await / async for / async with. Calling probe() → coroutine object; multi_probe() → async_generator; plain for → TypeError: not iterable.

@asynccontextmanager (3.7)

Same split as @contextmanager: before yield = __aenter__, after = __aexit__. Hattingh: await run_in_executor(None, download_webpage, url) then yield data then executor update_stats.

Native coro vs async gen

	native coro	async gen
async def	yes	yes
yield	never	required
return value	allowed	only empty return
driven by	await, create_task, …	async for, async comprehensions
awaitable?	yes	no

PEP 530 (3.6)

Async generator expression can be defined anywhere: (domain async for domain, found in multi_probe(names) if found). Use it only inside async def (or the asyncio REPL).

List/dict/set with async for / await only in async def (or REPL).

[i async for i in aiter() if i % 2] ≡ the obvious loop. [await fun() for fun in funcs] ≡ gather order (submit order if you sort names first). {name: found async for name, found in multi_probe(names)}. if (await probe(name)).found needs parens — . binds tighter than await.

Typing (prefer collections.abc on 3.9+)

AsyncContextManager, AsyncIterable, AsyncIterator, AsyncGenerator[Yield, Send], Awaitable, Coroutine[Yield, Send, Return]. Yield covariant; Send contravariant. AsyncGenerator has no return type (unlike Generator) — StopIteration hacks are not how native coros work.

Curio (and Trio)

async/await are language, not asyncio. Beazley rebuilt a loop after 3.4’s yield from asyncio. Cleaner: curio.socket.getaddrinfo (no loop object), run(main()) instead of get/run_until_complete/close.

TaskGroup: async with, await group.spawn(probe, domain), async for task in group: as tasks finish — as_completed unnecessary. g.results instead of gather. Structured concurrency: one enter/exit; all spawned work done or cancelled (like structured programming vs GOTO). Trio nurseries. PEP 654 exception groups / except* aimed at asyncio catching up.

Curio: spawn_thread, AWAIT(coro) from threads, UniversalQueue with asyncio in another thread. In 2021 few HTTP/DB drivers; strong low-level/WebSocket/Happy Eyeballs samples. asyncio later deprecated passing loop= everywhere.

How it works / doesn’t

Dahl: treating file/net I/O like x+1 is wrong. ~1 GHz: L1 “3 s” human scale → RAM minutes → disk years → net worse. Overlap waits or you bought a very expensive sleep.

There are I/O-bound functions, not I/O-bound systems. Under load something is CPU. Slow loop = every connection waits. Measure early. Options: process pool, task queue from day one, Cython/C/Rust (drop GIL), or accept the hit and write it down. Threads still GIL. First async def breeds more; blocking libs become the tax. Videla again.

Quiz

async gen vs coro

1 / 2

async def with yield in the body

Pick an answer to see if you are right.

Starter multi never probes; collect returns [].