파이썬 이터레이터와 컨텍스트 매니저 직접 구현 - 프로토콜과 with 원리 (파이썬 고급 17단원)
이 단원에서 배우는 것
7단원에서 with open(...) 으로 파일을 열었고, 11단원에서 yield 로 대용량 로그를 일정한 메모리로 훑었다. 둘 다 파이썬이 주는 문법을 쓰기만 한 것이다. 고급 과정의 첫 단원은 그 두 문법이 어떤 약속(프로토콜) 위에서 돌아가는지를 열어 보고, 그 약속을 우리 코드가 직접 구현한다. 제너레이터가 왜 두 번 못 도는지, with 가 예외를 어떻게 다루는지는 프로토콜을 봐야 답이 나온다. 이 단원부터 24단원까지는 하나의 소재를 계속 키운다. 액세스 로그를 읽어 느린 요청 비율을 감시하는 CLI 도구 logkit 이다. 여기서는 그 도구의 심장인 로그 리더를 만든다.
- 이터러블과 이터레이터를 구분하고
__iter__·__next__를 직접 구현한다 - 11단원의 제너레이터가 그 프로토콜을 어떻게 대신해 주는지 확인하고 파이프라인으로 조립한다
__enter__·__exit__와contextlib로 예외가 나도 자원이 반드시 정리되게 만든다
왜 필요한가
운영 서버의 하루치 액세스 로그는 흔히 수 GB 다. 여기서 응답이 1초 넘는 요청의 비율을 구해야 한다고 하자. 가장 먼저 떠오르는 코드는 이것이다.
lines = open("access.log", encoding="utf-8").readlines() # 전부 메모리로
entries = [parse(l) for l in lines] # 한 번 더 메모리로
slow = [e for e in entries if e.ms >= 1000] # 또 한 번4GB 로그면 이 코드는 파싱 결과까지 합쳐 10GB 이상을 요구한다. 서버에서 OOM 으로 죽거나, 죽지 않더라도 스왑에 들어가 몇십 분을 잡아먹는다. 게다가 첫 줄을 처리하기까지 파일 전체를 다 읽어야 하므로, 문제가 있는 로그를 만나 죽는 시점은 30분 뒤가 된다.
필요한 것은 한 줄씩 흘려보내는 구조다. 메모리는 한 줄치만 쓰고, 중간 단계를 몇 개를 붙이든 총 메모리는 늘지 않으며, 첫 결과가 즉시 나온다. 파이썬에서 이 구조의 이름이 이터레이터다.
그리고 흘려보내는 동안 파일 핸들, DB 커넥션, 락, 임시 디렉터리 같은 자원이 열려 있다. 중간에 예외가 나도 이것들은 반드시 닫혀야 한다. try/finally 를 호출하는 쪽마다 쓰는 대신, 정리 책임을 객체 안에 넣고 with 한 줄로 강제하는 것이 컨텍스트 매니저다. 이 두 프로토콜은 짝이다. 스트리밍을 하면 자원이 오래 열려 있고, 그래서 정리가 더 중요해진다.
문법과 예제
이터러블과 이터레이터는 다른 것이다
for x in obj 는 사실 이렇게 동작한다.
it = iter(obj) # obj.__iter__() 를 부른다 → 이터레이터를 얻는다
while True:
try:
x = next(it) # it.__next__() 를 부른다
except StopIteration:
break
...- 이터러블 —
__iter__가 있는 것. 리스트, 딕셔너리, 파일 객체. 몇 번이든 새로 순회할 수 있다. - 이터레이터 —
__next__가 있고__iter__는 자기 자신을 돌려주는 것. 한 번 쓰면 끝이다.
리스트를 두 번 돌 수 있고 제너레이터는 못 도는 이유가 여기 있다.
로그 리더를 이터레이터로 직접 구현하기
from __future__ import annotations
import re
from dataclasses import dataclass
LINE = re.compile(
r'^(?P<ip>\S+) \S+ \S+ \[(?P<ts>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+)[^"]*" '
r'(?P<status>\d{3}) (?P<bytes>\d+|-) (?P<ms>\d+)$'
)
@dataclass(frozen=True, slots=True)
class Entry:
ip: str
path: str
status: int
ms: int
class LogReader:
"""파일 하나를 받아 Entry 를 순서대로 내놓는 이터레이터."""
def __init__(self, path: str) -> None:
self._fp = open(path, encoding="utf-8")
self.bad = 0 # 파싱 실패 줄 수를 세어 둔다
def __iter__(self) -> "LogReader":
return self # 이터레이터는 자기 자신을 돌려준다
def __next__(self) -> Entry:
if self._fp.closed:
raise StopIteration # 소진된 뒤 또 불려도 StopIteration 을 유지한다
for raw in self._fp:
m = LINE.match(raw.rstrip("\n"))
if m is None:
self.bad += 1
continue
return Entry(m["ip"], m["path"], int(m["status"]), int(m["ms"]))
self._fp.close()
raise StopIteration # 끝났다는 신호. return 이 아니다__next__ 안에서 for 를 도는 것이 이상해 보일 수 있는데, 파일 객체 자체가 이터레이터이므로 self._fp 는 지난번에 멈춘 자리를 기억한다. 깨진 줄은 건너뛰고 다음 유효한 줄 하나만 돌려주는 구조다.
제너레이터로 다시 쓰면 절반 이하가 된다
11단원에서 쓴 yield 가 바로 위 프로토콜의 자동 구현이다. 정규식은 15단원의 명명 그룹을 그대로 쓴다.
from collections.abc import Iterable, Iterator
def read_log(path: str) -> Iterator[Entry]:
with open(path, encoding="utf-8") as fp:
for raw in fp:
m = LINE.match(raw.rstrip("\n"))
if m is not None:
yield Entry(m["ip"], m["path"], int(m["status"]), int(m["ms"]))
def slow_only(entries: Iterable[Entry], limit_ms: int) -> Iterator[Entry]:
return (e for e in entries if e.ms >= limit_ms)
def ratio(entries: Iterable[Entry], limit_ms: int) -> float:
total = slow = 0
for e in entries:
total += 1
slow += e.ms >= limit_ms # bool 은 int 다
return slow / total if total else 0.0yield 가 있는 함수를 호출하면 본문이 실행되지 않고 제너레이터 객체만 돌아온다. 첫 next() 에서 비로소 yield 까지 실행되고 거기서 멈춘다. 상태 저장(self._fp, self.bad)을 우리가 손으로 관리할 필요가 사라진 것이 핵심이다.
함수를 이렇게 나눠 놓으면 파이프라인으로 조립된다. 각 단계는 값을 쌓아 두지 않으므로 단계를 늘려도 메모리는 그대로다.
from itertools import islice
entries = read_log("access.log")
slow = slow_only(entries, 1000)
for e in islice(slow, 10): # 앞 10건만 보고 멈춘다
print(e.path, e.ms)islice 로 10건만 뽑으면 파일도 딱 그만큼만 읽힌다. 리스트로 만들었다면 4GB 를 다 읽은 뒤에 앞 10개를 잘랐을 것이다.
컨텍스트 매니저: 단계별 시간 측정기
with 는 __enter__ 와 __exit__ 두 메서드만 요구한다.
import time
from types import TracebackType
class Stage:
"""파이프라인 한 단계의 소요 시간을 재고, 실패해도 기록은 남긴다."""
def __init__(self, name: str, report: dict[str, float]) -> None:
self.name = name
self.report = report
def __enter__(self) -> "Stage":
self.started = time.perf_counter()
return self # as 뒤에 바인딩되는 값
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> bool:
self.report[self.name] = time.perf_counter() - self.started
return False # 예외를 삼키지 않는다
report: dict[str, float] = {}
with Stage("parse", report):
entries = list(read_log("access.log"))
try:
with Stage("upload", report):
raise RuntimeError("전송 실패")
except RuntimeError as e:
print("예외는 그대로 올라온다:", e)
print(report) # {'parse': 0.31..., 'upload': 1.2e-05} — 실패한 단계도 기록됨__exit__ 의 반환값이 이 프로토콜의 전부다. True 를 돌려주면 예외가 사라지고, False(또는 None)면 그대로 전파된다. 대부분의 경우 답은 False 다.
@contextmanager 로 짧게
import time
from contextlib import contextmanager
from collections.abc import Iterator
@contextmanager
def stage(name: str, report: dict[str, float]) -> Iterator[None]:
started = time.perf_counter()
try:
yield # 여기서 with 블록 본문이 실행된다
finally:
report[name] = time.perf_counter() - startedyield 앞이 __enter__, 뒤가 __exit__ 다. try/finally 를 빼먹으면 안 된다. with 블록에서 예외가 나면 그 예외가 yield 지점에서 다시 던져지므로, finally 가 없으면 정리 코드가 통째로 건너뛰어진다. 예외를 삼키고 싶다면 except 로 잡고 return 하면 되는데, 클래스 방식의 return True 와 달리 여기서는 실수로 삼키기가 더 어렵다.
개수가 정해지지 않은 자원: ExitStack
로그가 날짜별로 쪼개져 있어 파일 N 개를 동시에 열어야 한다면 with 를 몇 개 쓸지 알 수 없다. ExitStack 이 이 문제를 푼다.
from contextlib import ExitStack
from pathlib import Path
paths = sorted(Path("logs").glob("access-*.log"))
with ExitStack() as stack:
files = [stack.enter_context(p.open(encoding="utf-8")) for p in paths]
merged = (line for fp in files for line in fp)
print(sum(1 for _ in merged))
# 블록을 벗어나면 등록된 역순으로 전부 닫힌다. 중간에 예외가 나도 마찬가지다같은 contextlib 에서 자주 쓰는 것 둘을 덧붙인다.
import os
from contextlib import chdir, suppress
with suppress(FileNotFoundError): # try/except/pass 한 줄 대체
os.remove("logkit.tmp")
with chdir("/var/log/nginx"): # 3.11 에서 추가됐다
print(os.getcwd())
# 블록을 벗어나면 원래 디렉터리로 돌아온다
contextlib.chdir는 파이썬 3.11 부터 있다. 3.10 이하를 지원해야 한다면 직접 만들어야 한다. 그리고 프로세스의 현재 디렉터리는 스레드 전체가 공유하므로, 멀티스레드 코드에서는chdir자체를 피하고 절대 경로를 쓰는 편이 낫다.
ExitStack 의 고급 사용법과 비동기 컨텍스트 매니저(async with)는 contextlib 공식 문서에 정리돼 있다.
실무에서 자주 틀리는 것
1. 제너레이터를 두 번 순회하고 0 을 얻는다
entries = read_log("access.log")
total = sum(1 for _ in entries)
slow = sum(1 for e in entries if e.ms >= 1000) # 항상 0
print(slow / total) # 항상 0.0첫 sum 이 제너레이터를 끝까지 소진했으므로 두 번째는 빈 이터레이터를 돈다. 예외도 경고도 없이 조용히 0 이 나온다는 점이 고약하다. 리포트 수치가 0 으로 찍히는데 원인을 못 찾는 버그의 단골이다.
해결은 셋 중 하나다.
- 한 번의 순회로 두 값을 같이 센다(위
ratio함수처럼). 가장 좋다. - 함수가 아니라 호출 가능한 것을 넘겨서 필요할 때마다 새 제너레이터를 만든다.
- 정말 작은 데이터라면
entries = list(read_log(...))로 한 번만 실체화한다. 이건 메모리를 포기하는 결정이므로 의식적으로 해야 한다.
itertools.tee 로 갈래를 나눌 수도 있지만, 한쪽이 크게 앞서 나가면 그 차이만큼을 내부 버퍼에 쌓는다. 로그 스트리밍에서는 결국 전부 메모리에 올라가므로 답이 되지 않는다.
함수를 만들 때는 인자 타입을 list 가 아니라 Iterable 로 받되, 인자를 한 번만 순회하도록 구현하는 것을 규칙으로 삼는다. 두 번 돌아야 하는 함수라면 시그니처를 Sequence 로 바꿔서 호출자에게 알린다.
2. __exit__ 에서 실수로 True 를 돌려준다
def __exit__(self, exc_type, exc, tb):
self.report[self.name] = time.perf_counter() - self.started
return True # ← 로그만 남기려 했는데 예외가 통째로 사라진다파이썬 함수는 명시적으로 return 하지 않으면 None(거짓)을 돌려주므로 보통은 안전하다. 문제는 정리 로직 끝에 self.cleanup() 같은 호출의 결과를 무심코 return 하거나, "에러도 기록했으니 처리한 셈"이라고 판단해 True 를 넣는 경우다. 이렇게 되면 배치 작업이 실패했는데 종료 코드는 0 이고, 모니터링은 성공으로 집계한다. 예외를 정말 삼켜야 한다면 특정 타입만 골라서 삼킨다.
def __exit__(self, exc_type, exc, tb):
self.report[self.name] = time.perf_counter() - self.started
return exc_type is not None and issubclass(exc_type, TimeoutError)3. 제너레이터가 끝까지 소비되지 않으면 정리 시점이 미뤄진다
def read_log(path):
with open(path, encoding="utf-8") as fp: # 이 with 는 제너레이터 안에 있다
for raw in fp:
yield raw
def first_error(path):
for line in read_log(path):
if " 500 " in line:
return line # ← 여기서 함수를 나가면 제너레이터는 중간에 멈춘 채 남는다제너레이터가 yield 에서 멈춘 채 버려지면 with 의 종료는 그 객체가 가비지 컬렉션될 때 일어난다. CPython 은 참조 카운트가 0 이 되는 즉시 정리하므로 대개 티가 안 나지만, 순환 참조에 걸리거나 PyPy 같은 다른 구현에서는 파일 핸들이 한참 열려 있다. 로그 파일 수천 개를 도는 배치에서 Too many open files 로 터지는 전형적 경로다. 정리 시점을 확정하고 싶으면 호출하는 쪽에서 contextlib.closing 을 쓰거나 명시적으로 닫는다.
from contextlib import closing
with closing(read_log(path)) as lines: # 블록을 벗어날 때 gen.close() 가 불린다
for line in lines:
if " 500 " in line:
break4. __iter__ 에서 self 를 돌려주면 안 되는 경우가 있다
위 LogReader 는 파일 하나에 대응하는 일회용 객체라 return self 가 맞다. 하지만 "로그 디렉터리"처럼 여러 번 순회하는 게 자연스러운 컬렉션을 만들 때 같은 짓을 하면, 두 번째 for 부터 빈 결과가 나온다. 다시 돌 수 있어야 하는 객체는 __iter__ 가 호출될 때마다 새 이터레이터를 만들어 돌려줘야 한다.
class LogDir:
def __init__(self, root: str) -> None:
self.root = root
def __iter__(self): # 호출될 때마다 새 제너레이터
for path in sorted(Path(self.root).glob("access-*.log")):
yield from read_log(str(path))스스로 확인하기
read_log가 만든 제너레이터에len()을 쓰면TypeError가 난다. 왜 파이썬은 제너레이터에__len__을 주지 않았는가?- 아래
timed_open은 파일을 열고 소요 시간을 재려는 컨텍스트 매니저다. 버그가 두 개 있다. 찾아서 고쳐라.@contextmanager def timed_open(path): started = time.perf_counter() fp = open(path, encoding="utf-8") yield fp fp.close() print(time.perf_counter() - started) - 파일 목록을 받아 전체를 통틀어 가장 느린 요청 5건을 돌려주는 함수를 써라. 파일이 100개, 각 1GB 라고 가정한다. 메모리에 올려도 되는 것은 무엇이고 아닌 것은 무엇인가?
정답
- 길이를 알려면 끝까지 순회해야 하는데, 그러면 제너레이터가 소진되어 정작 값을 쓸 수 없다. 게다가 무한 제너레이터에는 길이 자체가 없다. "길이를 미리 안다"는 것은 데이터를 이미 다 갖고 있다는 뜻이므로 스트리밍의 전제와 모순된다. 개수가 필요하면 순회하면서 직접 세고, 그 순회에서 필요한 다른 집계도 같이 해야 한다.
- (1)
yield fp가try/finally밖에 있다. with 블록에서 예외가 나면fp.close()와 시간 출력이 실행되지 않는다. (2)fp.close()를 직접 부르는 대신 파일 객체 자체를with로 감싸는 편이 안전하다. 고친 코드:@contextmanager def timed_open(path): started = time.perf_counter() try: with open(path, encoding="utf-8") as fp: yield fp finally: print(time.perf_counter() - started) - 엔트리 전체를 모으면 안 되고, 상위 5건만 유지하면 된다.
heapq.nlargest가 정확히 이 일을 한다. 내부적으로 크기 5짜리 힙만 들고 스트림을 한 번 훑으므로 메모리는 파일 개수와 무관하게 일정하다.import heapq from itertools import chain def top_slow(paths: list[str], n: int = 5) -> list[Entry]: stream = chain.from_iterable(read_log(p) for p in paths) return heapq.nlargest(n, stream, key=lambda e: e.ms)sorted(...)[:5]로 쓰면 전체를 리스트로 만든 뒤 정렬하므로 메모리가 터진다.chain.from_iterable은 제너레이터 표현식을 받으므로 파일도 한 번에 하나씩만 열린다.
다음 단원에서는 이 파이프라인을 여러 파일에 동시에 돌린다. 그 순간 GIL 이라는 벽을 만나고, 스레드와 프로세스 중 무엇을 골라야 하는지가 문제가 된다.