Merge branch '2026-generators'

Author: egges

2026/generators/00_basic.py

@@ -0,0 +1,21 @@
+from typing import Generator
+
+
+def read_logs() -> Generator[str, None, None]:
+    lines = [
+        "info User logged in",
+        "warning Slow database query",
+        "error Payment failed",
+    ]
+    for line in lines:
+        print(f"producing: {line}")
+        yield line
+
+
+def main():
+    for line in read_logs():
+        print(f"consuming: {line}")
+
+
+if __name__ == "__main__":
+    main()

2026/generators/01_structured.py

@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from enum import StrEnum, auto
+from typing import Generator, Iterable
+
+
+# --- Domain model ---
+class LogLevel(StrEnum):
+    INFO = auto()
+    WARNING = auto()
+    ERROR = auto()
+
+
+@dataclass(slots=True)
+class LogRecord:
+    level: LogLevel
+    message: str
+
+
+# --- Source generator ---
+def read_logs() -> Generator[str, None, None]:
+    lines = [
+        "info User logged in",
+        "warning Slow database query",
+        "error Payment failed",
+    ]
+    for line in lines:
+        print(f"producing: {line}")
+        yield line
+
+
+# --- Transformation step ---
+def parse_logs(lines: Iterable[str]) -> Generator[LogRecord, None, None]:
+    for line in lines:
+        level_text, message = line.split(" ", maxsplit=1)
+        level = LogLevel(level_text)  # validated conversion
+        yield LogRecord(level=level, message=message)
+
+
+def handle_records(records: Iterable[LogRecord]) -> None:
+    for record in records:
+        print(f"handling: {record}")
+
+
+# --- Application entry point ---
+def main() -> None:
+    records = parse_logs(read_logs())
+    handle_records(records)
+
+
+if __name__ == "__main__":
+    main()

2026/generators/02_composition.py

@@ -0,0 +1,86 @@
+from dataclasses import dataclass
+from enum import StrEnum, auto
+from functools import reduce
+from typing import Any, Callable, Generator, Iterable
+
+
+# --- Domain model ---
+class LogLevel(StrEnum):
+    INFO = auto()
+    WARNING = auto()
+    ERROR = auto()
+
+
+@dataclass(slots=True)
+class LogRecord:
+    level: LogLevel
+    message: str
+
+
+# --- Source generator ---
+def read_logs() -> Generator[str, None, None]:
+    lines = [
+        "info User logged in",
+        "warning Slow database query",
+        "error Payment failed",
+    ]
+    for line in lines:
+        print(f"producing: {line}")
+        yield line
+
+
+# --- Pipeline stages ---
+def parse_logs(lines: Iterable[str]) -> Generator[LogRecord, None, None]:
+    for line in lines:
+        level_text, message = line.split(" ", maxsplit=1)
+        level = LogLevel(level_text)
+        yield LogRecord(level=level, message=message)
+
+
+def filter_important(
+    records: Iterable[LogRecord],
+) -> Generator[LogRecord, None, None]:
+    for record in records:
+        if record.level in {LogLevel.WARNING, LogLevel.ERROR}:
+            yield record
+
+
+def normalize_messages(
+    records: Iterable[LogRecord],
+) -> Generator[LogRecord, None, None]:
+    for record in records:
+        yield LogRecord(
+            level=record.level,
+            message=record.message.lower(),
+        )
+
+
+# --- Composition helper ---
+type PipelineStage = Callable[[Iterable[Any]], Iterable[Any]]
+
+
+def compose(*stages: PipelineStage) -> PipelineStage:
+    def apply(data: Iterable[Any]) -> Iterable[Any]:
+        return reduce(lambda acc, stage: stage(acc), stages, data)
+
+    return apply
+
+
+def handle_records(records: Iterable[LogRecord]) -> None:
+    for record in records:
+        print(f"handling: {record}")
+
+
+# --- Application entry point ---
+def main() -> None:
+    pipeline = compose(
+        parse_logs,
+        filter_important,
+        normalize_messages,
+    )
+
+    handle_records(pipeline(read_logs()))
+
+
+if __name__ == "__main__":
+    main()

2026/generators/03_backpressure.py

@@ -0,0 +1,92 @@
+import time
+from dataclasses import dataclass
+from enum import StrEnum, auto
+from functools import reduce
+from typing import Any, Callable, Generator, Iterable
+
+
+# --- Domain model ---
+class LogLevel(StrEnum):
+    INFO = auto()
+    WARNING = auto()
+    ERROR = auto()
+
+
+@dataclass(slots=True)
+class LogRecord:
+    level: LogLevel
+    message: str
+
+
+# --- Source generator ---
+def read_logs() -> Generator[str, None, None]:
+    lines = [
+        "info User logged in",
+        "warning Slow database query",
+        "error Payment failed",
+    ]
+    for line in lines:
+        print(f"producing: {line}")
+        yield line
+
+
+# --- Pipeline stages ---
+def parse_logs(lines: Iterable[str]) -> Generator[LogRecord, None, None]:
+    for line in lines:
+        print(f"parsing: {line}")
+        level_text, message = line.split(" ", maxsplit=1)
+        level = LogLevel(level_text)
+        yield LogRecord(level=level, message=message)
+
+
+def filter_important(
+    records: Iterable[LogRecord],
+) -> Generator[LogRecord, None, None]:
+    for record in records:
+        print(f"filtering: {record}")
+        if record.level in {LogLevel.WARNING, LogLevel.ERROR}:
+            yield record
+
+
+def normalize_messages(
+    records: Iterable[LogRecord],
+) -> Generator[LogRecord, None, None]:
+    for record in records:
+        print(f"normalizing: {record}")
+        yield LogRecord(
+            level=record.level,
+            message=record.message.lower(),
+        )
+
+
+# --- Composition helper ---
+type PipelineStage = Callable[[Iterable[Any]], Iterable[Any]]
+
+
+def compose(*stages: PipelineStage) -> PipelineStage:
+    def apply(data: Iterable[Any]) -> Iterable[Any]:
+        return reduce(lambda acc, stage: stage(acc), stages, data)
+
+    return apply
+
+
+# --- Slow consumer ---
+def handle_records(records: Iterable[LogRecord]) -> None:
+    for record in records:
+        print(f"handling: {record}")
+        time.sleep(1)
+
+
+# --- Application entry point ---
+def main() -> None:
+    pipeline = compose(
+        parse_logs,
+        filter_important,
+        normalize_messages,
+    )
+
+    handle_records(pipeline(read_logs()))
+
+
+if __name__ == "__main__":
+    main()

2026/generators/04_send.py

@@ -0,0 +1,70 @@
+from dataclasses import dataclass
+from enum import StrEnum, auto
+from typing import Generator, Iterable
+
+
+# --- Domain model ---
+class LogLevel(StrEnum):
+    INFO = auto()
+    WARNING = auto()
+    ERROR = auto()
+
+
+@dataclass(slots=True)
+class LogRecord:
+    level: LogLevel
+    message: str
+
+
+# --- Source generator ---
+def read_logs() -> Generator[str, None, None]:
+    lines = [
+        "info User logged in",
+        "warning Slow database query",
+        "error Payment failed",
+        "warning Disk space low",
+    ]
+    for line in lines:
+        print(f"producing: {line}")
+        yield line
+
+
+def parse_logs(lines: Iterable[str]) -> Generator[LogRecord, None, None]:
+    for line in lines:
+        level_text, message = line.split(" ", maxsplit=1)
+        yield LogRecord(level=LogLevel(level_text), message=message)
+
+
+def should_emit(record: LogRecord, threshold: LogLevel) -> bool:
+    if threshold is LogLevel.WARNING:
+        return record.level in {LogLevel.WARNING, LogLevel.ERROR}
+    return record.level is LogLevel.ERROR
+
+
+# --- Simpler send() example ---
+def threshold_filter() -> Generator[LogLevel, LogLevel | None, None]:
+    threshold = LogLevel.ERROR
+    while True:
+        new_threshold = yield threshold
+        if new_threshold is not None:
+            threshold = new_threshold
+
+
+def main() -> None:
+    records = parse_logs(read_logs())
+
+    filter_settings = threshold_filter()
+    threshold = next(filter_settings)  # prime the generator
+
+    for index, record in enumerate(records):
+        if index == 2:
+            threshold = filter_settings.send(LogLevel.WARNING)
+        else:
+            threshold = next(filter_settings)
+
+        if should_emit(record, threshold):
+            print(f"handling: {record}")
+
+
+if __name__ == "__main__":
+    main()

2026/generators/04b_return.py

@@ -0,0 +1,50 @@
+from collections.abc import Generator, Iterable
+from dataclasses import dataclass
+from enum import StrEnum, auto
+
+
+class LogLevel(StrEnum):
+    INFO = auto()
+    WARNING = auto()
+    ERROR = auto()
+
+
+@dataclass(slots=True)
+class LogRecord:
+    level: LogLevel
+    message: str
+
+
+def read_logs() -> Generator[str, None, None]:
+    lines = [
+        "info User logged in",
+        "warning Slow database query",
+        "error Payment failed",
+    ]
+    for line in lines:
+        yield line
+
+
+def parse_logs(lines: Iterable[str]) -> Generator[LogRecord, None, int]:
+    count = 0
+    for line in lines:
+        level_text, message = line.split(" ", maxsplit=1)
+        yield LogRecord(level=LogLevel(level_text), message=message)
+        count += 1
+
+    return count  # 👈 final result
+
+
+def main() -> None:
+    parser = parse_logs(read_logs())
+
+    try:
+        while True:
+            record = next(parser)
+            print(f"handling: {record}")
+    except StopIteration as e:
+        print(f"Total records processed: {e.value}")
+
+
+if __name__ == "__main__":
+    main()