Python 데코레이터·제너레이터·컨텍스트 조합 7패턴

서버/데이터 파이프라인을 Python으로 만들다 보면 세 가지 도구를 자주 만납니다.

데코레이터: 횡단 관심사(로깅, 권한, 재시도, 캐시)를 함수에 얹는다.
제너레이터: 스트리밍 처리로 메모리 사용을 줄이고 파이프라인을 단순화한다.
컨텍스트 매니저: 리소스(파일, 락, 트랜잭션, 스팬)를 안전하게 열고 닫는다.

각각만 써도 좋지만, 현실에서는 조합이 성능/안정성/가독성을 좌우합니다. 이 글은 세 요소를 섞어 쓰는 대표 패턴 7가지를 코드 중심으로 정리합니다.

참고로 “에러를 예외로만 처리하지 않는” 사고방식은 파이프라인 안정성에 도움이 됩니다. 다른 언어 사례지만 C++23 std expected로 예외 없는 에러 처리도 같이 보면 설계 감이 잡힙니다.

준비: 표준 라이브러리만으로 조합하기

Python에서는 contextlib 하나로 조합의 대부분이 해결됩니다.

@contextmanager: 제너레이터를 컨텍스트 매니저로 변환
ContextDecorator: 컨텍스트 매니저를 데코레이터로도 사용
ExitStack: 동적 리소스 묶음 관리

아래 패턴들은 가능한 한 표준 라이브러리로만 구성합니다.

패턴 1) 컨텍스트 매니저를 데코레이터로 승격하기

목표: with 와 @decorator 를 동일한 구현으로 재사용.

contextlib.ContextDecorator 를 상속하면 컨텍스트 매니저를 그대로 데코레이터로 쓸 수 있습니다.

from contextlib import ContextDecorator
import time

class timed(ContextDecorator):
    def __init__(self, name: str = ""):
        self.name = name

    def __enter__(self):
        self.t0 = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc, tb):
        dt = time.perf_counter() - self.t0
        label = self.name or "block"
        print(f"[timed] {label}: {dt:.6f}s")
        return False  # 예외는 삼키지 않음

@timed("work")
def work(n: int) -> int:
    s = 0
    for i in range(n):
        s += i
    return s

with timed("manual"):
    work(100_000)

장점

동일한 코드로 블록/함수 모두 계측 가능
테스트 시에도 with timed(): 로 범위를 좁혀 검증 가능

주의

데코레이터로 쓸 때도 __exit__ 의 반환값(예외 삼킴 여부)이 그대로 적용됩니다.

패턴 2) `@contextmanager` 로 “안전한 임시 상태” 만들기

목표: 전역/스레드 로컬/환경변수/설정 값을 잠깐 바꾸고 반드시 복구.

제너레이터 기반 컨텍스트 매니저가 가장 읽기 쉽습니다.

from contextlib import contextmanager
import os

@contextmanager
def temp_environ(**updates):
    old = os.environ.copy()
    os.environ.update({k: str(v) for k, v in updates.items()})
    try:
        yield
    finally:
        os.environ.clear()
        os.environ.update(old)

with temp_environ(LOG_LEVEL="DEBUG", FEATURE_X=1):
    # 이 범위에서만 환경변수 변경
    ...

조합 포인트

이 컨텍스트 매니저는 패턴 1과 결합해 데코레이터로도 쉽게 승격할 수 있습니다(필요하면 ContextDecorator 로 래핑).

패턴 3) 제너레이터 파이프라인에 “측정/로깅”을 데코레이터로 주입

목표: 스트리밍 처리에서 각 단계의 처리량/지연을 관찰.

제너레이터 함수는 반환이 아니라 yield 로 흘러가기 때문에, 데코레이터가 “입출력 스트림”을 감싸는 형태가 됩니다.

from collections.abc import Iterable, Iterator
import time

def log_throughput(name: str):
    def deco(gen_fn):
        def wrapper(*args, **kwargs) -> Iterator:
            t0 = time.perf_counter()
            n = 0
            for item in gen_fn(*args, **kwargs):
                n += 1
                yield item
            dt = time.perf_counter() - t0
            rps = n / dt if dt else float("inf")
            print(f"[pipe] {name}: {n} items in {dt:.3f}s ({rps:.1f} items/s)")
        return wrapper
    return deco

@log_throughput("filter_even")
def filter_even(xs: Iterable[int]) -> Iterator[int]:
    for x in xs:
        if x % 2 == 0:
            yield x

@log_throughput("square")
def square(xs: Iterable[int]) -> Iterator[int]:
    for x in xs:
        yield x * x

data = range(1_000_00)
result = square(filter_even(data))
# 소비 시점에 실제 실행
print(next(result))

핵심

제너레이터는 “소비될 때 실행”되므로, 측정도 소비 완료 시점에 출력됩니다.
중간 단계에 로깅을 넣어도 메모리 사용량은 증가하지 않습니다(스트리밍 유지).

패턴 4) 컨텍스트 매니저로 “제너레이터 클린업” 보장하기

목표: 스트리밍 중간에 소비자가 중단해도 파일/소켓/커서가 닫히게 하기.

제너레이터 내부에서 with open(...) 을 쓰면 대부분 해결되지만, “리소스를 열고 제너레이터를 반환”해야 하는 구조라면 contextmanager 로 감싸는 편이 안전합니다.

from contextlib import contextmanager
from collections.abc import Iterator

@contextmanager
def iter_lines(path: str) -> Iterator[Iterator[str]]:
    f = open(path, "r", encoding="utf-8")
    try:
        def gen() -> Iterator[str]:
            for line in f:
                yield line.rstrip("\n")
        yield gen()
    finally:
        f.close()

with iter_lines("app.log") as lines:
    for line in lines:
        if "ERROR" in line:
            print(line)
            break  # 중간 중단해도 close 보장

왜 이런 패턴이 필요한가

제너레이터만 반환하면, 소비자가 끝까지 돌지 않을 때 close() 타이밍이 애매해질 수 있습니다.
컨텍스트 매니저가 “범위”를 제공하므로 리소스 생명주기가 명확해집니다.

패턴 5) `ExitStack` 으로 동적 리소스 묶기 + 파이프라인에 주입

목표: 입력이 몇 개일지 모를 때 파일/락/임시디렉터리 등을 안전하게 다루기.

ExitStack 은 런타임에 컨텍스트 매니저를 조건부로 쌓고, 마지막에 역순으로 정리합니다.

from contextlib import ExitStack
from collections.abc import Iterator

def merge_files(paths: list[str]) -> Iterator[str]:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p, "r", encoding="utf-8")) for p in paths]
        for f in files:
            for line in f:
                yield line

for line in merge_files(["a.txt", "b.txt", "c.txt"]):
    ...

조합 포인트

이 패턴을 쓰면 “동적으로 열린 리소스”가 제너레이터의 생명주기와 함께 안전하게 정리됩니다.
대규모 스트리밍/배치 처리에서는 메모리뿐 아니라 파일 디스크립터 누수가 장애 원인이 되므로 특히 중요합니다.

패턴 6) “트랜잭션/락”을 컨텍스트로, “재시도/백오프”를 데코레이터로

목표: DB/분산락/파일락 등 임계구역을 안전하게 감싸고, 실패 시 재시도 정책을 분리.

컨텍스트 매니저는 “획득/해제”를 담당하고, 데코레이터는 “정책”을 담당하게 쪼개면 유지보수가 쉬워집니다.

import time
from contextlib import contextmanager

class TransientError(RuntimeError):
    pass

def retry(max_attempts: int, base_sleep: float = 0.1):
    def deco(fn):
        def wrapper(*args, **kwargs):
            attempt = 0
            while True:
                try:
                    return fn(*args, **kwargs)
                except TransientError:
                    attempt += 1
                    if attempt >= max_attempts:
                        raise
                    time.sleep(base_sleep * (2 ** (attempt - 1)))
        return wrapper
    return deco

@contextmanager
def fake_lock(name: str):
    print(f"lock acquire: {name}")
    try:
        yield
    finally:
        print(f"lock release: {name}")

@retry(max_attempts=3)
def critical_update():
    with fake_lock("order:123"):
        # 여기서 일시 오류가 나면 재시도
        raise TransientError("temporary")

try:
    critical_update()
except TransientError:
    print("failed after retries")

실전 팁

재시도는 멱등성(idempotency)과 세트로 고려해야 합니다.
락/트랜잭션을 어디 범위에 걸지(재시도마다 새로 잡을지, 한 번 잡고 유지할지)는 장애 양상에 큰 영향을 줍니다.

패턴 7) 컨텍스트 매니저로 “관측(Tracing/Span)” 범위 만들고, 제너레이터 단계별로 중첩

목표: 스트리밍 파이프라인에서 단계별 지연을 추적 가능한 구조로 만들기.

분산 트레이싱 라이브러리를 쓰든, 자체 로그를 남기든 핵심은 동일합니다. “단계 범위”를 컨텍스트로 만들고, 제너레이터 각 단계에서 중첩 컨텍스트를 엽니다.

from contextlib import contextmanager
import time
from collections.abc import Iterable, Iterator

@contextmanager
def span(name: str):
    t0 = time.perf_counter()
    print(f"[span start] {name}")
    try:
        yield
    finally:
        dt = time.perf_counter() - t0
        print(f"[span end] {name}: {dt:.4f}s")

def stage(name: str):
    def deco(gen_fn):
        def wrapper(xs: Iterable[int]) -> Iterator[int]:
            with span(name):
                for item in gen_fn(xs):
                    yield item
        return wrapper
    return deco

@stage("parse")
def parse(xs: Iterable[int]) -> Iterator[int]:
    for x in xs:
        yield int(x)

@stage("compute")
def compute(xs: Iterable[int]) -> Iterator[int]:
    for x in xs:
        yield x * 2

@stage("sink")
def sink(xs: Iterable[int]) -> Iterator[int]:
    for x in xs:
        # 실제로는 저장/전송 같은 부작용이 위치
        yield x

pipeline = sink(compute(parse(range(10000))))
# 소비가 시작될 때 span이 열리고, 소비가 끝날 때 닫힘
for _ in pipeline:
    pass

왜 이게 중요한가

파이프라인 병목은 “어느 단계가 느린지”가 핵심입니다.
단계별 span을 만들면, 추후 OpenTelemetry 같은 구현으로 옮겨도 구조를 그대로 유지할 수 있습니다.

관련해서 “폭주를 제어하는 가드레일” 사고방식은 LLM 에이전트에서도 동일하게 중요합니다. 패턴 6의 재시도/백오프/차단은 LangChain 에이전트 무한루프·툴콜 폭주 차단법과도 연결됩니다.

조합 패턴을 고를 때 체크리스트

1) 범위가 먼저다: 블록이냐, 함수냐, 스트림이냐

블록 범위: 컨텍스트 매니저가 가장 명확
함수 범위: 데코레이터가 가장 간결
스트림 범위: 제너레이터 + “소비 시점”을 고려한 계측 필요

2) 실패 모델을 분리하라

리소스 정리 실패, 일시 오류, 영구 오류를 한 덩어리로 처리하면 재시도/롤백/로그가 꼬입니다.
예외 기반이든 결과 기반이든(예: Result 류), “정책(데코레이터)”과 “범위(컨텍스트)”를 분리하면 훨씬 단단해집니다.

3) 성능은 대체로 “메모리”와 “대기”에서 터진다

제너레이터는 메모리 문제를 줄이지만, 대기(락/IO/네트워크)는 줄여주지 않습니다.
대기/병목을 찾으려면 패턴 1, 3, 7 같은 관측 도구를 먼저 붙이세요.

대기와 병목을 진단하는 접근은 인프라에서도 같습니다. 예를 들어 네트워크 단에서 “Pod는 뜨는데 트래픽이 0” 같은 상황은 관측 지점과 범위 설정이 핵심인데, 이는 EKS Pod는 뜨는데 트래픽 0 - NetPol·SG·CNI 10분 진단 같은 글과도 결이 같습니다.

마무리: 7가지 패턴을 한 문장으로 요약

ContextDecorator 로 컨텍스트를 데코레이터로 승격
@contextmanager 로 임시 상태를 안전하게 변경/복구
제너레이터 데코레이터로 파이프라인 관측(처리량/지연)
컨텍스트로 제너레이터 클린업 보장
ExitStack 으로 동적 리소스 묶기
컨텍스트는 임계구역, 데코레이터는 재시도 정책
span 컨텍스트를 단계별 제너레이터에 중첩해 트레이싱 구조화

이 조합들을 습관화하면 “코드가 짧아지는 것”보다 더 큰 이득을 얻습니다. 장애 시 원인 추적이 쉬워지고, 리소스 누수/중복 작업/재시도 폭주 같은 운영 이슈를 구조적으로 줄일 수 있습니다.