import asyncio: Learn Python's AsyncIO #5 - Batteries Included

Main ideas & lessons (structured)

1) AsyncIO course direction and “language-level” async constructs

• This is the 5th episode in a beginner series introducing Python asyncio.
• The episode promises coverage of three core async language protocols:
  • Asynchronous context managers (async with)
  • Asynchronous iterators (async for)
  • Asynchronous generators (advanced, but very useful)
• These are then applied to a real-world example showing how asyncio benefits:
  • Networking
  • Interprocess communication (conceptually/task cooperation)
  • Cooperative task scheduling within a single thread using coroutines

---

2) Context managers: why async with exists

Blocking version (baseline)

• A database client example demonstrates:
  • Connect using a DSN
  • Define a user type/schema (name, optional local date)
  • Insert using parameterized queries (positional args)
  • Select data back
  • Disconnect/close afterward
• Problem: manually disconnect/close is error-prone and can leave dangling connections, wasting resources.

Synchronous context manager pattern (Python with)

• Uses a context manager with:
  • __enter__: establish connection
  • __exit__: close connection (and return whether exceptions were handled)
• The with block ensures cleanup on exit—even if exceptions occur.
• Limitation: in the database example, these operations are blocking, causing latency and preventing non-blocking concurrency.

Asynchronous context manager pattern (Python async with)

• Reason for change:
  • __enter__ / __exit__ are regular functions, so you can’t directly use await inside them.
• Solution:
  • Use async context managers with async-capable equivalents (aenter/aexit conceptually).
  • Now connection establishment/close can be awaited using async IO methods.
• Result:
  • Same logic/output as blocking, but now supports concurrency via coroutines.

---

3) Connection pooling and proper resource lifecycle

• Creating new database connections is expensive, so:
  • Create a pool once in an application entry point.
  • Pass the pool into user code later.
• In user code:
  • async with pool.acquire() to get a connection safely.
  • Await queries inside the block.
• When the outer context manager ends (e.g., server shutdown), it ensures:
  • Pool shutdown
  • Connections released/closed cleanly

---

4) Async iteration: async for and why anext exists

Normal iteration recap

• Iteration protocols use:
  • __iter__ → returns an iterator
  • __next__ → returns next item or raises StopIteration
• But for IO:
  • read operations are blocking and should be awaited.

Async iteration protocol

• Async equivalent transforms:
  • iter → aiter
  • next → anext (or “async next” behavior)
  • StopIteration → StopAsyncIteration
• Key nuance:
  • aiter is not necessarily async def, because it mainly needs to produce a valid async iterator; IO setup can occur during the first anext call.

Real-world use case: streaming large query results

• Example uses asyncpg with PostgreSQL:
  • Create a pool
  • Acquire a connection
  • Use a cursor for streaming
  • Iterate with async for record in con.cursor(...)
• Benefits:
  • Lower latency to first result
  • Lower memory usage (doesn’t fetch everything at once)

Under the hood (cursor buffering concept)

• The cursor uses an async iterator with buffering (FIFO queue/buffer):
  • Fetches multiple records at once internally
  • Yields them one by one to the consumer
• Benefit: users still see incremental streaming while implementation remains efficient.

---

5) Async generators: generator-based coroutines with cleanup (aclose)

• The example introduces an async generator:
  • An async def function that uses yield.
  • It takes an argument: another async function read_line.
  • It repeatedly:
    • awaits read_line()
    • yields values until the returned value is empty
• Type modeling:
  • The generator function returns an async iterator.
  • yielded values are typed (example: bytes).

Partial iteration and cleanup hazard

• Async generators can be used in async for loops.
• If a loop is broken early:
  • cleanup may need to run (e.g., commit/rollback transactions, release connections)
• Concern: if not explicitly closed, cleanup might not run promptly.

The aclose() mechanism and asyncio hooks

• When aclose() is called (and awaited), it:
  • interrupts the async generator
  • triggers generator-exit flow so cleanup executes
• Even if you don’t call it manually:
  • CPython/asyncio track async generators via “async gen hooks”
  • cleanup is scheduled when generators are garbage collected
• Therefore, resource finalization is handled safely by the event loop/runtime.

---

6) Real-world asyncio application: text chat server/client

A staged evolution demonstrates practical async features.

Stage 1: Minimal echo server/client

• Client:
  • Establishes a network connection via asyncio
  • Reads from input and sends bytes line-by-line
  • Uses stream reader/writer:
    • reading is awaited
    • writing requires writer.drain() to flush buffers
• Server:
  • async handler reads chunks and echoes back
  • Runs “serve forever” style loop

Stage 2: Simulate network latency + break the server (byte-by-byte issue)

• Client sends one byte at a time with delays.
• Problem:
  • Server treats each byte as a separate message
  • special quit may fail due to abrupt end handling

Stage 3: Fix framing using async generators on the server

• Server read loop changes from simple reads to:
  • async for message in splitlines_async_generator(reader): ...
• This async generator:
  • accumulates chunks
  • detects newline boundaries
  • yields complete messages
• Result:
  • quit works again
  • message boundaries become correct under latency

Stage 4: Server-initiated greeting needs concurrency

• Issue:
  • server greeting wasn’t received until client started sending input
• Fix:
  • client uses a concurrent task for reading/writing so it can receive while waiting for user input
• Additional fix:
  • replace blocking stdin/file reads with async-capable IO using aiofiles
  • otherwise the event loop stalls

Stage 5: Introduce nicknames and routing messages

• Server supports commands:
  • @ nickname: message recipient
  • IM / introduce command: identify nickname (no real authentication)
• Server maintains a mapping:
  • nickname -> write_soon (earlier approach)
• Problem:
  • multiple concurrent writers to the same recipient can interleave (“world salad”)

Stage 6: Fix message ordering with per-user queues (final architecture)

• Server stores for each user:
  • a queue and a single consumer task that writes messages to that user
• Design:
  • Many producers put messages into the queue
  • Exactly one consumer gets messages and writes them out
• Message shutdown signal:
  • an empty message indicates the writer should shut down
• Client refactoring:
  • send-file function becomes a manager spawning concurrent tasks:
    • network reader
    • network writer
    • file copier
  • uses task cancellation and return_when style completion detection:
    • stop when file ends OR quit occurs OR shutdown signal is received
• Server refactoring:
  • command handling moved into handle_commands
  • uses try/finally to ensure:
    • disconnected users removed
    • connection closed gracefully
• Outcome:
  • concurrent messaging is serialized per recipient
  • quit behaves correctly

---

7) Synchronization primitives in asyncio (locks/events/semaphores/conditions)

The video closes by outlining built-in primitives and intended uses.

When to use them

• Use when threading-like coordination is needed:
  • shared mutable data between coroutines
  • correct ordering (e.g., server must be ready before a resource connection)
• Warning:
  • concurrency primitives are tricky; race conditions and deadlocks are possible
• asyncio is primarily for non-blocking IO; heavy computation can block the loop unless offloaded.

Methods and intended usage (detailed list)

• Lock

  • One coroutine acquires; others wait.
  • Use the async with lock: pattern.
  • lock.locked() exists mostly for logging; polling isn’t atomic.
  • No explicit timeout parameters in shown APIs; prefer asyncio.wait_for wrapping instead.

• Event

  • Communication via a boolean-like flag:
    • one coroutine calls “set” to signal
    • others call await event.wait() to wait
  • clear() unsets the event.
  • event.set() is typically not awaitable; waiting is awaitable.
  • Not an async context manager (nothing to “return” on exit).

• Semaphore

  • Like a lock, but allows up to N concurrent holders.
  • Use to limit load on expensive/restricted resources.
  • value determines the max concurrent acquisitions.

• Condition

  • More complex two-sided producer/consumer coordination:
    • producer holds condition/lock, modifies shared state, then calls notify/notify_all
    • consumer holds condition/lock and waits via await condition.wait()
  • Use when a lock alone isn’t enough to know whether a resource is usable.
  • Allows notifying multiple waiting consumers.

---

8) Plans for next episode

• Next episode will focus on more integrated applications rather than encyclopedic summaries.
• Mentioned example:
  • a Starlette web application
  • using HDB as the database
• Encouragement to subscribe.

---

Speakers / sources featured

• Lukasz (speaker/host): “Hi, this is Lukasz from HDB…”
• AsyncIO / CPython source code referenced:
  • asyncio internals, especially CPython’s base_events.py (e.g., run_forever, async gen hooks)
• Libraries/tools mentioned:
  • HDB
  • asyncpg
  • aiofiles
  • contextlib
  • asyncio constructs (queues, tasks, synchronization primitives)
  • Starlette