feat: five modern-Python patterns — catalog complete at 32 units

patterns/modern/__init__.py

@@ -0,0 +1 @@
+"""Modern Python patterns beyond the Gang of Four."""

patterns/modern/async_producer_consumer/README.md

@@ -0,0 +1,44 @@
+---
+id: modern/async_producer_consumer
+name: Async Producer/Consumer
+aliases: [asyncio-queue, worker-pool, pipeline]
+guide_url: null
+problem: "Decouple work generation from work processing under asyncio, with bounded memory and clean shutdown."
+symptoms: ["fan out downloads to workers", "bounded queue backpressure", "asyncio pipeline", "graceful worker shutdown"]
+verdict: use-with-care
+caveats:
+  - "Choose one shutdown discipline and test it: sentinels per worker, or queue.join() plus task cancellation."
+  - "An unbounded queue turns a slow consumer into a memory leak — set maxsize and let backpressure work."
+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.

patterns/modern/async_producer_consumer/__init__.py

@@ -0,0 +1 @@
+"""Async Producer/Consumer: bounded queues between async workers."""

patterns/modern/async_producer_consumer/naive.py

@@ -0,0 +1,39 @@
+"""The thread version: queue.Queue, sentinel-per-worker shutdown.
+
+Works, but every worker is an OS thread and the coordination is manual.
+"""
+
+from __future__ import annotations
+
+import queue
+import threading
+
+
+def process_all(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
+    for w in workers:
+        w.join()
+    return sorted(results)
+
+
+def main() -> None:
+    print(process_all(["a", "b", "c", "d"]))
+
+
+if __name__ == "__main__":
+    main()

patterns/modern/async_producer_consumer/pythonic.py

@@ -0,0 +1,41 @@
+"""asyncio.Queue + TaskGroup workers.
+
+maxsize bounds memory (backpressure), join() waits for all items to be
+processed, cancellation ends the idle workers.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+
+async def process_all(items: list[str], worker_count: int = 3) -> list[str]:
+    channel: asyncio.Queue[str] = asyncio.Queue(maxsize=2)  # backpressure
+    results: list[str] = []
+
+    async def worker() -> None:
+        while True:
+            item = await channel.get()
+            try:
+                await asyncio.sleep(0)  # stand-in for real async I/O
+                results.append(item.upper())
+            finally:
+                channel.task_done()
+
+    async with asyncio.TaskGroup() as group:
+        workers = [group.create_task(worker()) for _ in range(worker_count)]
+        for item in items:
+            await channel.put(item)  # blocks when the queue is full
+        await channel.join()  # all items fetched AND task_done()
+        for w in workers:
+            w.cancel()  # idle workers end; TaskGroup absorbs the cancellation
+
+    return sorted(results)
+
+
+def main() -> None:
+    print(asyncio.run(process_all(["a", "b", "c", "d", "e"])))
+
+
+if __name__ == "__main__":
+    main()

patterns/modern/async_producer_consumer/real_world.py

@@ -0,0 +1,49 @@
+"""The idiom shaped as a pipeline: N workers, bounded queue, ordered results.
+
+A fake fetcher stands in for HTTP so the demo and tests run offline; swap it
+for a real client and nothing else changes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+
+Fetcher = Callable[[str], Awaitable[str]]
+
+
+async def fake_fetch(url: str) -> str:
+    await asyncio.sleep(0)
+    return f"body-of-{url}"
+
+
+async def crawl(urls: list[str], fetch: Fetcher = fake_fetch, workers: int = 4) -> dict[str, str]:
+    """Fan URLs out to workers; collect {url: body} whatever the finish order."""
+    channel: asyncio.Queue[str] = asyncio.Queue(maxsize=8)
+    pages: dict[str, str] = {}
+
+    async def worker() -> None:
+        while True:
+            url = await channel.get()
+            try:
+                pages[url] = await fetch(url)
+            finally:
+                channel.task_done()
+
+    async with asyncio.TaskGroup() as group:
+        tasks = [group.create_task(worker()) for _ in range(workers)]
+        for url in urls:
+            await channel.put(url)
+        await channel.join()
+        for t in tasks:
+            t.cancel()
+    return pages
+
+
+def main() -> None:
+    urls = [f"https://example.com/{n}" for n in range(3)]
+    print(asyncio.run(crawl(urls)))
+
+
+if __name__ == "__main__":
+    main()

patterns/modern/async_producer_consumer/tests/test_async_producer_consumer.py

@@ -0,0 +1,36 @@
+"""Behavioral tests for all three producer/consumer variants."""
+
+from patterns.modern.async_producer_consumer import naive, pythonic, real_world
+
+
+class TestNaive:
+    def test_thread_pool_processes_everything(self) -> None:
+        assert naive.process_all(["a", "b", "c", "d"]) == ["A", "B", "C", "D"]
+
+    def test_zero_items(self) -> None:
+        assert naive.process_all([]) == []
+
+
+class TestPythonic:
+    async def test_all_items_processed_despite_backpressure(self) -> None:
+        items = [chr(ord("a") + n) for n in range(10)]  # more items than maxsize
+        assert await pythonic.process_all(items) == [c.upper() for c in items]
+
+    async def test_more_workers_than_items(self) -> None:
+        assert await pythonic.process_all(["x"], worker_count=5) == ["X"]
+
+    async def test_zero_items_shuts_down_cleanly(self) -> None:
+        assert await pythonic.process_all([]) == []
+
+
+class TestRealWorld:
+    async def test_crawl_collects_every_url(self) -> None:
+        urls = [f"u{n}" for n in range(9)]
+        pages = await real_world.crawl(urls, workers=3)
+        assert pages == {u: f"body-of-{u}" for u in urls}
+
+    async def test_injected_fetcher(self) -> None:
+        async def fetch(url: str) -> str:
+            return url[::-1]
+
+        assert await real_world.crawl(["abc"], fetch=fetch) == {"abc": "cba"}

patterns/modern/context_manager/README.md

@@ -0,0 +1,44 @@
+---
+id: modern/context_manager
+name: Context Manager
+aliases: [with-statement, RAII, resource-management]
+guide_url: null
+problem: "Guarantee acquire/release pairing around a block of code, even when it raises."
+symptoms: ["forgot to close", "cleanup on exception", "try/finally everywhere", "temporary state that must be restored"]
+verdict: pythonic
+caveats:
+  - "@contextlib.contextmanager wants the yield inside try/finally — without it, an exception in the body skips your cleanup."
+  - "Returning True from __exit__ swallows the exception; do it only on purpose."
+stdlib_sightings: [open, contextlib.contextmanager, contextlib.ExitStack, tempfile.TemporaryDirectory]
+---
+
+# Context Manager
+
+## Problem
+
+Every acquired resource — file, lock, connection, temporary state — must be
+released on *every* exit path. Hand-written `try/finally` scattered through a
+codebase is where cleanup bugs live.
+
+## Naive solution
+
+`naive.py` is the try/finally discipline done by hand, including the nested
+two-resource version that shows why it doesn't scale.
+
+## Pythonic solution
+
+The `with` statement makes the pairing structural: `pythonic.py` implements
+the protocol both ways — a class with `__enter__`/`__exit__`, and the
+generator form via `@contextmanager` where the `yield` splits acquire from
+release.
+
+## In the wild
+
+`open`, locks, and sqlite transactions are all context managers;
+`contextlib.ExitStack` manages a *dynamic* number of them, unwinding in
+reverse on the way out — shown in `real_world.py`.
+
+## Verdict
+
+**Pythonic.** Python's own RAII; any acquire/release pair you write twice
+deserves one.

patterns/modern/context_manager/__init__.py

@@ -0,0 +1 @@
+"""Context Manager: structural acquire/release pairing."""

patterns/modern/context_manager/naive.py

@@ -0,0 +1,55 @@
+"""Cleanup by hand: try/finally on every exit path.
+
+Correct -- and it must be re-written correctly at every call site.
+The nested version shows why the discipline doesn't scale.
+"""
+
+from __future__ import annotations
+
+
+class Resource:
+    def __init__(self, name: str, log: list[str]) -> None:
+        self.name = name
+        self.log = log
+        self.log.append(f"open {name}")
+
+    def close(self) -> None:
+        self.log.append(f"close {self.name}")
+
+
+def use_one(log: list[str], *, explode: bool = False) -> None:
+    resource = Resource("a", log)
+    try:
+        log.append("work")
+        if explode:
+            raise RuntimeError("boom")
+    finally:
+        resource.close()
+
+
+def use_two(log: list[str]) -> None:
+    first = Resource("a", log)
+    try:
+        second = Resource("b", log)  # every extra resource nests another level
+        try:
+            log.append("work")
+        finally:
+            second.close()
+    finally:
+        first.close()
+
+
+def main() -> None:
+    import contextlib
+
+    log: list[str] = []
+    with contextlib.suppress(RuntimeError):  # itself a context manager!
+        use_one(log, explode=True)
+    print(f"cleanup survived the exception: {log}")
+    log.clear()
+    use_two(log)
+    print(f"nested by hand: {log}")
+
+
+if __name__ == "__main__":
+    main()

patterns/modern/context_manager/pythonic.py

@@ -0,0 +1,52 @@
+"""The protocol, both ways.
+
+A class with __enter__/__exit__, and the generator form where the yield is
+the seam between acquire and release. Note the try/finally around the yield:
+without it, an exception in the body skips cleanup.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+from types import TracebackType
+
+
+class Managed:
+    def __init__(self, name: str, log: list[str]) -> None:
+        self.name = name
+        self.log = log
+
+    def __enter__(self) -> Managed:
+        self.log.append(f"open {self.name}")
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        self.log.append(f"close {self.name}")  # returning None: never swallow
+
+
+@contextmanager
+def managed(name: str, log: list[str]) -> Iterator[str]:
+    log.append(f"open {name}")
+    try:
+        yield name
+    finally:
+        log.append(f"close {name}")
+
+
+def main() -> None:
+    log: list[str] = []
+    with Managed("a", log):
+        log.append("work")
+    with managed("b", log):
+        log.append("more work")
+    print(log)
+
+
+if __name__ == "__main__":
+    main()