yield from+send로 코루틴 파이프라인 구현하기

from __future__ import annotations
from typing import Generator, Iterable

def coroutine(func):
    """코루틴을 자동 프라이밍하는 데코레이터."""
    def wrapper(*args, **kwargs):
        g = func(*args, **kwargs)
        next(g)  # prime
        return g
    return wrapper

@coroutine
def printer(prefix: str = "") -> Generator[None, str, None]:
    try:
        while True:
            item = yield
            print(f"{prefix}{item}")
    except GeneratorExit:
        # close() 호출 시 정리 로직
        # 파일 flush, 네트워크 종료 같은 작업을 여기에 둘 수 있음
        return

p = printer("OUT: ")
p.send("hello")
p.send("world")
p.close()

from typing import Callable, Generator, TypeVar

T = TypeVar("T")
U = TypeVar("U")

@coroutine
def c_filter(pred: Callable[[T], bool], target) -> Generator[None, T, None]:
    try:
        while True:
            item = yield
            if pred(item):
                target.send(item)
    except GeneratorExit:
        target.close()
        return

@coroutine
def c_map(fn: Callable[[T], U], target) -> Generator[None, T, None]:
    try:
        while True:
            item = yield
            target.send(fn(item))
    except GeneratorExit:
        target.close()
        return

sink = printer("RESULT: ")
stage = c_filter(lambda s: "ERROR" in s, c_map(str.lower, sink))

stage.send("INFO boot")
stage.send("ERROR Disk Full")
stage.send("ERROR Network")
stage.close()

from typing import Generator, Callable

@coroutine
def receiver(handler: Callable[[object], None]) -> Generator[None, object, None]:
    try:
        while True:
            item = yield
            handler(item)
    except GeneratorExit:
        return

@coroutine
def map_stage(fn, target):
    def handle(item):
        target.send(fn(item))
    try:
        # receiver 코루틴에 모든 send/throw/close를 위임
        yield from receiver(handle)
    finally:
        target.close()

@coroutine
def filter_stage(pred, target):
    def handle(item):
        if pred(item):
            target.send(item)
    try:
        yield from receiver(handle)
    finally:
        target.close()

pipe = filter_stage(lambda s: s.startswith("#"),
                   map_stage(lambda s: s.strip(), printer("TAG: ")))

pipe.send(" hello ")
pipe.send("#  topic ")
pipe.send("#python ")
pipe.close()

from typing import Iterable

def drive(source: Iterable[str], target) -> None:
    try:
        for item in source:
            target.send(item)
    finally:
        target.close()

lines = ["INFO boot", "ERROR A", "WARN x", "ERROR B"]
pipe = c_filter(lambda s: s.startswith("ERROR"), printer("ALERT: "))

drive(lines, pipe)

from collections import deque

@coroutine
def batcher(n: int, target):
    buf = []
    try:
        while True:
            item = yield
            buf.append(item)
            if len(buf) >= n:
                target.send(list(buf))
                buf.clear()
    except GeneratorExit:
        # 남은 버퍼 flush
        if buf:
            target.send(list(buf))
        target.close()
        return

@coroutine
def list_printer(prefix: str = ""):
    try:
        while True:
            items = yield
            print(prefix + ",".join(map(str, items)))
    except GeneratorExit:
        return

pipe = batcher(3, list_printer("BATCH: "))
for i in range(8):
    pipe.send(i)
pipe.close()

@coroutine
def broadcast(*targets):
    try:
        while True:
            item = yield
            for t in targets:
                t.send(item)
    except GeneratorExit:
        for t in targets:
            t.close()
        return

pipe = broadcast(
    c_filter(lambda s: "ERROR" in s, printer("E: ")),
    c_filter(lambda s: "WARN" in s, printer("W: ")),
)

drive(["INFO a", "WARN b", "ERROR c", "WARN d"], pipe)

class BadRecord(ValueError):
    pass

@coroutine
def validator(target):
    try:
        while True:
            item = yield
            if not isinstance(item, str) or not item:
                raise BadRecord("empty or non-string")
            target.send(item)
    except GeneratorExit:
        target.close()
        return

pipe = validator(printer("OK: "))
pipe.send("hello")
try:
    pipe.send("")
except BadRecord as e:
    print("caught:", e)
finally:
    pipe.close()

import re

LOG_RE = re.compile(r"level=(?P<level>\w+)\s+msg=(?P<msg>.+)")

@coroutine
def parse_log(target):
    try:
        while True:
            line = yield
            m = LOG_RE.search(line)
            if not m:
                continue
            target.send({"level": m.group("level"), "msg": m.group("msg")})
    except GeneratorExit:
        target.close()
        return

@coroutine
def pick_error(target):
    try:
        while True:
            rec = yield
            if rec.get("level") == "ERROR":
                target.send(rec)
    except GeneratorExit:
        target.close()
        return

@coroutine
def normalize_msg(target):
    try:
        while True:
            rec = yield
            rec = dict(rec)
            rec["msg"] = rec["msg"].lower().strip()
            target.send(rec)
    except GeneratorExit:
        target.close()
        return

@coroutine
def print_batch(prefix: str = ""):
    try:
        while True:
            batch = yield
            print(prefix + " | ".join(r["msg"] for r in batch))
    except GeneratorExit:
        return

pipeline = parse_log(
    pick_error(
        normalize_msg(
            batcher(2, print_batch("ERR-BATCH: "))
        )
    )

lines = [
    "level=INFO msg=Boot",
    "level=ERROR msg=Disk Full ",
    "level=ERROR msg=Network Down",
    "level=WARN msg=Retry",
    "level=ERROR msg=Permission Denied",
]

drive(lines, pipeline)