feat: v2 modules — modern group

patterns/modern/async_producer_consumer/README.md

@@ -14,31 +14,17 @@ stdlib_sightings: [asyncio.Queue, asyncio.TaskGroup, queue.Queue]
 
 # Async Producer/Consumer
 
-## Problem
-
-Producers generate work faster (or slower) than consumers process it. You
-want N workers pulling from a shared source, bounded memory in between, and
-a shutdown that neither drops items nor hangs.
-
-## Naive solution
-
-`naive.py` is the thread version: `threading.Thread` workers around a
-`queue.Queue` with sentinels — fine, but each worker burns an OS thread and
-coordination is manual.
-
-## Pythonic solution
-
-`asyncio.Queue` with `TaskGroup`-managed workers: `maxsize` gives
-backpressure, `queue.join()` waits for completion, cancellation ends the
-idle workers. All the coordination is in the queue.
-
-## In the wild
-
-This *is* the stdlib idiom — the asyncio docs' own queue example is this
-pattern; `real_world.py` shapes it as a rate-limited fetch pipeline with
-per-item results collected in completion order.
-
-## Verdict
-
-**Use with care.** The right tool for I/O-bound fan-out; get the shutdown
-discipline right (and tested) or debug it forever.
+Fan I/O-bound work out to N workers over a bounded queue — backpressure by
+`maxsize`, shutdown as an explicit, tested choice. **Verdict: use with care**
+— the right tool for async fan-out; the two caveats are where it bites.
+
+| Where | What |
+|---|---|
+| [`pattern/`](pattern/) | The importable code: `WorkerPool`, `Shutdown`, `process_all` |
+| [`docs/`](docs/) | [Fundamentals](docs/fundamentals.md) · [Implementation guide](docs/implementation.md) · [External examples](docs/examples.md) |
+| [`examples/feed_fetcher/`](examples/feed_fetcher/) | Mini-project: feed pipeline with per-item failure capture, both shutdown disciplines |
+| [`tests/`](tests/) | Behavioral tests for the pool and the mini-project |
+
+```bash
+uv run python -m patterns.modern.async_producer_consumer.examples.feed_fetcher
+```

patterns/modern/async_producer_consumer/__init__.py

@@ -1 +1,13 @@
-"""Async Producer/Consumer: bounded queues between async workers."""
+"""Async Producer/Consumer — public API.
+
+>>> from patterns.modern.async_producer_consumer import WorkerPool
+"""
+
+from patterns.modern.async_producer_consumer.pattern import (
+    Processor,
+    Shutdown,
+    WorkerPool,
+    process_all,
+)
+
+__all__ = ["Processor", "Shutdown", "WorkerPool", "process_all"]

patterns/modern/async_producer_consumer/docs/examples.md

@@ -0,0 +1,25 @@
+# Async Producer/Consumer — external examples
+
+Real embodiments of the pattern outside this repo, for deeper study.
+
+## Standard library
+
+- **`asyncio.Queue`** — the buffer itself; the docs include a worked
+  producer/consumer example that is this pattern verbatim.
+  <https://docs.python.org/3/library/asyncio-queue.html>
+- **`asyncio.TaskGroup`** (3.11+) — structured lifetime for the worker
+  tasks; the reason the pool needs no manual join/cancel bookkeeping
+  beyond its shutdown discipline.
+  <https://docs.python.org/3/library/asyncio-task.html#task-groups>
+- **`queue.Queue`** — the threaded flavor, with the same
+  `task_done()`/`join()` contract the JOIN_AND_CANCEL discipline uses.
+  <https://docs.python.org/3/library/queue.html>
+- **`concurrent.futures`** — the pool-shaped alternative when items are
+  independent and you want futures rather than a shared queue.
+  <https://docs.python.org/3/library/concurrent.futures.html>
+
+## Elsewhere
+
+- **aiohttp** client examples — crawler-style fan-out over a session is
+  this pattern with real HTTP in the processor seam. *(unverified)*
+  <https://docs.aiohttp.org/>

patterns/modern/async_producer_consumer/docs/fundamentals.md

@@ -0,0 +1,86 @@
+# Async Producer/Consumer — fundamentals
+
+## Intent
+
+Decouple work *generation* from work *processing*: producers enqueue, N
+consumers dequeue and process, and a bounded buffer between them keeps a fast
+side from drowning a slow one. Under asyncio the pattern is how you fan
+I/O-bound work out to concurrent workers with bounded memory and a shutdown
+that neither drops items nor hangs.
+
+## Participants
+
+| Role | Classic (threaded) form | asyncio form |
+|---|---|---|
+| Buffer | `queue.Queue` + locks/conditions | `asyncio.Queue(maxsize=...)` — backpressure built in |
+| Producers | Threads calling `put` | Any coroutine calling `await queue.put(item)` |
+| Consumers | Worker threads in a `get` loop | N worker tasks — `WorkerPool` in [`pattern/pool.py`](../pattern/pool.py) |
+| Worker lifetime | Manual `start`/`join` | `asyncio.TaskGroup` owns the tasks structurally |
+| Shutdown discipline | Ad hoc, often forgotten | An explicit choice: `Shutdown.SENTINEL` or `Shutdown.JOIN_AND_CANCEL` |
+
+## Mechanism
+
+1. A bounded queue is created; `maxsize` is the memory budget *and* the
+   backpressure valve — `put` blocks when the buffer is full.
+2. N workers start, each looping `get → process`.
+3. Producers enqueue items; slow consumers automatically slow the producers.
+4. Shutdown, the part first attempts get wrong, is one of two disciplines:
+   - **Sentinel** — after the last item, enqueue one end-marker per worker;
+     each worker exits on dequeuing one.
+   - **Join and cancel** — workers mark `task_done()`; the coordinator awaits
+     `queue.join()` (every item fetched *and* finished), then cancels the
+     now-idle workers.
+
+## The classic form, and what Python absorbs
+
+Before asyncio this was the thread pattern — an OS thread per worker, a lock
+around shared results, and hand-rolled sentinel plumbing:
+
+```python
+def process_all_threaded(items: list[str], worker_count: int = 2) -> list[str]:
+    channel: queue.Queue[str | None] = queue.Queue()
+    results: list[str] = []
+    lock = threading.Lock()
+
+    def worker() -> None:
+        while (item := channel.get()) is not None:
+            with lock:
+                results.append(item.upper())
+
+    workers = [threading.Thread(target=worker) for _ in range(worker_count)]
+    for w in workers:
+        w.start()
+    for item in items:
+        channel.put(item)
+    for _ in workers:
+        channel.put(None)  # one sentinel per worker — forget one and it hangs
+    for w in workers:
+        w.join()
+    return sorted(results)
+```
+
+`asyncio.Queue` absorbs the locking entirely and `TaskGroup` absorbs the
+lifetime bookkeeping. What Python does *not* absorb is the design: choosing
+`maxsize`, and choosing — then testing — the shutdown discipline. That
+remainder is the pattern.
+
+## When to use it
+
+- Many I/O-bound items (fetches, uploads, API calls) and you want bounded
+  concurrency rather than a task per item.
+- Producers and consumers run at different, varying speeds and you need
+  memory to stay bounded in between.
+
+## When not to use it
+
+- CPU-bound work — an event loop serializes it; use a process pool.
+- Independent items with no need to bound in-flight memory —
+  `asyncio.gather` (or `TaskGroup` alone) over per-item tasks is simpler.
+- One item, one consumer — that is just an awaited call.
+
+## Verdict: use with care
+
+The right tool for I/O fan-out, with two sharp edges the caveats name: an
+unbounded queue turns a slow consumer into a memory leak, and an untested
+shutdown path is where these systems hang. `WorkerPool` makes both choices
+explicit arguments so they cannot be forgotten — only wrong on purpose.

patterns/modern/async_producer_consumer/docs/implementation.md

@@ -0,0 +1,48 @@
+# Async Producer/Consumer — implementation guide
+
+## The smell that calls for it
+
+An `async` code path does `for item in items: await do(item)` and the wall
+clock shows it — sequential awaits over independent I/O. Or the opposite:
+`gather` over ten thousand tasks and memory shows *that*.
+
+## Introducing it, step by step
+
+1. **Isolate the per-item coroutine.** One `async def process(item) -> result`
+   with no shared state. This is the seam everything else plugs into.
+2. **Pick the memory budget.** `maxsize` is how many items may sit fetched-
+   but-unprocessed. Small (2–16) is almost always right; it exists to create
+   backpressure, not to be a cache.
+3. **Pick the worker count.** Concurrency toward the slow resource — for
+   HTTP this is "how many connections is polite", not "how many items exist".
+4. **Pick the shutdown discipline — and write the test the same day.**
+   - `Shutdown.JOIN_AND_CANCEL` when a coordinator knows the item set and
+     wants "everything finished" as a joinable event.
+   - `Shutdown.SENTINEL` when workers should drain the queue and stop:
+     the pool enqueues one end-marker per worker after the items are
+     exhausted.
+5. **Decide the failure policy at the edge.** The pool is fail-fast (one bad
+   item cancels the run, surfacing as an `ExceptionGroup`). If a bad item
+   must not kill the batch, catch inside *your* processor and return an
+   outcome object — as the [feed_fetcher example](../examples/feed_fetcher/)
+   does with `FetchOutcome`.
+
+## Idioms
+
+- `async with asyncio.TaskGroup()` owns the workers; nothing outlives the
+  block, and worker exceptions propagate instead of vanishing.
+- Results in completion order, ordering as the caller's last step — sorting
+  inside the pool would hide the concurrency it exists to provide.
+- The same shape works threaded (`queue.Queue`, `concurrent.futures`) when
+  the work is blocking rather than async; the design choices carry over.
+
+## Pitfalls
+
+- **Unbounded queue.** `asyncio.Queue()` with no `maxsize` removes the
+  backpressure that is half the pattern's point.
+- **Mixed disciplines.** `task_done()` bookkeeping *and* sentinels in the
+  same pool — each looks redundant, together they deadlock or exit early.
+- **Sentinel miscounting.** Fewer end-markers than workers hangs the rest;
+  route all shutdown through one tested code path (the pool), not call sites.
+- **CPU-bound processors.** The event loop runs one coroutine at a time;
+  workers give you interleaved I/O waits, never parallel computation.

patterns/modern/async_producer_consumer/examples/__init__.py

@@ -0,0 +1 @@
+"""Mini-projects demonstrating the async producer/consumer pattern in practice."""

patterns/modern/async_producer_consumer/examples/feed_fetcher/__init__.py

@@ -0,0 +1,16 @@
+"""Feed-fetching pipeline built on the async producer/consumer pool.
+
+Run it: ``uv run python -m patterns.modern.async_producer_consumer.examples.feed_fetcher``
+"""
+
+from patterns.modern.async_producer_consumer.examples.feed_fetcher.fetcher import (
+    fetch_all,
+    fetch_entries,
+    summarize,
+)
+from patterns.modern.async_producer_consumer.examples.feed_fetcher.models import (
+    Feed,
+    FetchOutcome,
+)
+
+__all__ = ["Feed", "FetchOutcome", "fetch_all", "fetch_entries", "summarize"]

patterns/modern/async_producer_consumer/examples/feed_fetcher/__main__.py

@@ -0,0 +1,29 @@
+"""Demo: a batch of feeds through the pool, under both shutdown disciplines."""
+
+from __future__ import annotations
+
+import asyncio
+
+from patterns.modern.async_producer_consumer.examples.feed_fetcher.fetcher import (
+    fetch_all,
+    summarize,
+)
+from patterns.modern.async_producer_consumer.examples.feed_fetcher.models import Feed
+from patterns.modern.async_producer_consumer.pattern import Shutdown
+
+
+def main() -> None:
+    feeds = [
+        Feed("python-insider", "https://feeds.example/python-insider"),
+        Feed("lwn", "https://feeds.example/lwn"),
+        Feed("hn", "https://feeds.example/hn"),
+        Feed("dead-blog", "https://unreachable.example/rss"),
+        Feed("release-notes", "https://feeds.example/releases"),
+    ]
+    for shutdown in Shutdown:
+        outcomes = asyncio.run(fetch_all(feeds, shutdown=shutdown))
+        print(f"{shutdown.value}: {summarize(outcomes)}")
+
+
+if __name__ == "__main__":
+    main()

patterns/modern/async_producer_consumer/examples/feed_fetcher/fetcher.py

@@ -0,0 +1,57 @@
+"""The pipeline: N workers pull feeds through a bounded queue.
+
+A fake network stands in for HTTP so the demo and tests run offline and
+deterministically; swap ``fetch_entries`` for a real client and nothing
+else changes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Iterable
+
+from patterns.modern.async_producer_consumer.examples.feed_fetcher.models import (
+    Feed,
+    FetchOutcome,
+)
+from patterns.modern.async_producer_consumer.pattern import Shutdown, WorkerPool
+
+
+async def fetch_entries(feed: Feed) -> int:
+    """Pretend to fetch and parse one feed; raise on a bad host."""
+    await asyncio.sleep(0)  # stand-in for real async I/O
+    if "unreachable" in feed.url:
+        raise ConnectionError(f"cannot reach {feed.url}")
+    return len(feed.name)  # deterministic stand-in for "entries parsed"
+
+
+async def fetch_all(
+    feeds: Iterable[Feed],
+    *,
+    workers: int = 3,
+    maxsize: int = 2,
+    shutdown: Shutdown = Shutdown.JOIN_AND_CANCEL,
+) -> list[FetchOutcome]:
+    """Fetch every feed; failures become recorded outcomes, not crashes."""
+
+    async def capture(feed: Feed) -> FetchOutcome:
+        try:
+            return FetchOutcome(feed, entries=await fetch_entries(feed))
+        except ConnectionError as exc:
+            return FetchOutcome(feed, error=str(exc))
+
+    pool: WorkerPool[Feed, FetchOutcome] = WorkerPool(
+        capture, workers=workers, maxsize=maxsize, shutdown=shutdown
+    )
+    return await pool.run(feeds)
+
+
+def summarize(outcomes: Iterable[FetchOutcome]) -> str:
+    """One line a human can read at the end of a run."""
+    outcomes = list(outcomes)
+    fetched = sum(o.entries for o in outcomes if o.ok)
+    failed = [o.feed.name for o in outcomes if not o.ok]
+    line = f"{fetched} entries from {sum(o.ok for o in outcomes)} feeds"
+    if failed:
+        line += f"; failed: {', '.join(sorted(failed))}"
+    return line

patterns/modern/async_producer_consumer/examples/feed_fetcher/models.py

@@ -0,0 +1,30 @@
+"""Domain objects for the feed-fetching pipeline."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Feed:
+    """A feed to fetch: a name and its URL."""
+
+    name: str
+    url: str
+
+
+@dataclass(frozen=True)
+class FetchOutcome:
+    """What happened to one feed — success with entries, or a recorded error.
+
+    The pool itself is fail-fast; capturing per-feed failures is this
+    project's policy, applied inside its processor.
+    """
+
+    feed: Feed
+    entries: int = 0
+    error: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.error is None