[파이썬 100강] 79강. contextlib로 리소스 정리와 전처리-후처리 패턴 표준화하기
파일을 열고 닫거나, DB 세션을 시작하고 종료하거나, 임시 설정을 적용했다가 원복하는 코드는 거의 모든 서비스 코드에 반복해서 등장합니다. 문제는 이 패턴을 매번 try/finally로 직접 쓰다 보면 누락이 생기고, 예외가 났을 때 정리가 불완전해지기 쉽다는 점입니다. 이번 강의에서는 contextlib로 이런 전처리-후처리 패턴을 안전하게 묶는 방법을 바로 예제로 정리합니다.
핵심 개념
- 컨텍스트 매니저는
with블록의 진입 시점과 종료 시점에 실행할 로직을 강제해, 자원 정리를 자동화합니다. contextlib.contextmanager데코레이터를 쓰면 클래스 없이도 제너레이터 한 함수로 컨텍스트 매니저를 만들 수 있습니다.ExitStack은 “몇 개가 열릴지 런타임에 결정되는 자원”을 한곳에서 관리할 때 가장 실무적인 선택지입니다.
with 문은 문법이 단순해 보이지만, 실제로는 안정성 도구입니다. 운영 환경에서는 파일 핸들, 소켓, 락, 트랜잭션 같은 자원이 닫히지 않으면 장애가 누적됩니다. 특히 예외가 중간에 터질 때 cleanup 경로가 빠지는 문제가 자주 생깁니다. contextlib는 이런 정리 책임을 코드 구조로 강제합니다.
또 하나 중요한 포인트는 “관심사 분리”입니다. 본문 비즈니스 로직에서는 핵심 작업만 읽히고, 부수적인 준비/정리 코드는 컨텍스트로 숨길 수 있습니다. 그래서 코드 리뷰할 때도 "무엇을 하는지"와 "실패 시 어떻게 복구되는지"가 동시에 명확해집니다.
기본 사용
예제 1) contextmanager로 가장 단순한 전처리-후처리
>>> from contextlib import contextmanager
>>>
>>> @contextmanager
... def timer(label: str):
... import time
... start = time.perf_counter()
... print(f"[START] {label}")
... try:
... yield
... finally:
... end = time.perf_counter()
... print(f"[END] {label} took {end - start:.4f}s")
...
>>> with timer("sum-loop"):
... total = sum(range(100000))
... print(total)
...
[START] sum-loop
4999950000
[END] sum-loop took 0.00xxs
해설:
yield이전은 진입 로직, 이후는 종료 로직입니다.- 블록 내부에서 예외가 나도
finally가 실행되어 종료 처리가 보장됩니다. - 로그/측정/락 해제처럼 "무조건 실행되어야 하는 뒷정리"를 함수 하나로 표준화할 수 있습니다.
예제 2) 예외를 기록하고 다시 던지는 패턴
>>> from contextlib import contextmanager
>>>
>>> @contextmanager
... def error_boundary(task_name: str):
... print(f"run={task_name}")
... try:
... yield
... except Exception as e:
... print(f"[ERROR] {task_name}: {type(e).__name__}: {e}")
... raise
... finally:
... print(f"cleanup={task_name}")
...
>>> with error_boundary("division"):
... 10 / 2
...
run=division
cleanup=division
>>> with error_boundary("broken"):
... 10 / 0
...
run=broken
[ERROR] broken: ZeroDivisionError: division by zero
cleanup=broken
Traceback (most recent call last):
...
ZeroDivisionError: division by zero
해설:
- 예외를 삼켜 버리지 않고 기록 후
raise로 재전파하는 게 핵심입니다. - 이렇게 하면 상위 계층의 실패 처리 정책(재시도, 알림, 중단)을 깨지 않습니다.
- 실무에서는 트랜잭션, 배치 단계, 외부 API 호출 단위를 감싸는 데 자주 씁니다.
예제 3) ExitStack으로 동적 자원 개수 안전하게 관리
>>> from contextlib import ExitStack
>>> from pathlib import Path
>>>
>>> p1 = Path("a.txt"); p2 = Path("b.txt")
>>> _ = p1.write_text("hello\n", encoding="utf-8")
>>> _ = p2.write_text("python\n", encoding="utf-8")
>>>
>>> files = ["a.txt", "b.txt"]
>>> with ExitStack() as stack:
... handles = [stack.enter_context(open(name, "r", encoding="utf-8")) for name in files]
... lines = [h.readline().strip() for h in handles]
... print(lines)
...
['hello', 'python']
해설:
- 파일 개수가 고정이면
with open(...) as f1, open(...) as f2로 충분하지만,
개수가 가변이면ExitStack이 훨씬 깔끔합니다. - 중간에 하나라도 실패하면 이미 열린 핸들은
ExitStack이 알아서 닫아 줍니다. - 다중 연결(파일, 네트워크 세션, 락)을 일괄 관리할 때 운영 안정성 차이가 크게 납니다.
예제 4) 임시 설정 적용 후 자동 복원
>>> from contextlib import contextmanager
>>>
>>> CONFIG = {"debug": False, "timeout": 3}
>>>
>>> @contextmanager
... def temporary_config(**patch):
... backup = CONFIG.copy()
... CONFIG.update(patch)
... try:
... yield CONFIG
... finally:
... CONFIG.clear()
... CONFIG.update(backup)
...
>>> print(CONFIG)
{'debug': False, 'timeout': 3}
>>> with temporary_config(debug=True, timeout=10) as c:
... print(c)
...
{'debug': True, 'timeout': 10}
>>> print(CONFIG)
{'debug': False, 'timeout': 3}
해설:
- 테스트에서 전역 설정을 바꿀 때 특히 유용합니다.
- 원복 로직을 강제하지 않으면 테스트 순서에 따라 결과가 흔들리는 문제가 생깁니다.
- "잠깐 바꾸고 반드시 되돌린다"는 규칙을 코드로 보장할 수 있습니다.
자주 하는 실수
실수 1) contextmanager 함수에서 yield를 빼먹음
>>> from contextlib import contextmanager
>>>
>>> @contextmanager
... def broken_cm():
... print("enter")
... # yield 누락
... print("exit")
...
>>> with broken_cm():
... pass
...
Traceback (most recent call last):
...
TypeError: 'NoneType' object is not an iterator
원인:
@contextmanager는 제너레이터 기반으로 동작합니다.yield가 없으면 컨텍스트 매니저 규약을 만족하지 못합니다.
해결:
>>> @contextmanager
... def fixed_cm():
... print("enter")
... try:
... yield
... finally:
... print("exit")
...
>>> with fixed_cm():
... print("work")
...
enter
work
exit
실수 2) 예외를 로그만 찍고 삼켜 버림
- 증상: 실패했는데도 상위 호출자는 성공으로 인식해 데이터 불일치가 누적됨
- 원인:
except Exception: ...에서raise를 하지 않음 - 해결: 정말 의도적으로 무시하는 예외가 아니면 반드시 재전파하고, 무시가 필요하면 주석/근거를 남겨 팀 규칙으로 관리
실수 3) with 블록 밖에서 이미 닫힌 자원을 계속 사용
>>> with open("a.txt", "r", encoding="utf-8") as f:
... first = f.readline().strip()
...
>>> f.readline()
Traceback (most recent call last):
...
ValueError: I/O operation on closed file.
원인:
- 컨텍스트 종료 시점에 자원이 닫힌다는 사실을 놓침
해결:
>>> with open("a.txt", "r", encoding="utf-8") as f:
... content = f.read()
...
>>> content.splitlines()[0]
'hello'
실수 4) ExitStack 대신 중첩 with를 과도하게 늘림
- 증상:
with가 5~6단 중첩되어 본문 로직이 안 보임 - 원인: 자원 개수/조합이 가변인데 고정 패턴으로만 작성
- 해결: 가변 자원은
ExitStack, 고정 소수 자원은 일반with로 구분해 가독성과 안전성 둘 다 확보
실무 패턴
-
입력 검증 규칙
- 컨텍스트 매니저 내부에서 사용할 경로, 타임아웃, 모드 값은 진입 전에 검증합니다.
- 진입 이후 실패는 "외부 환경 문제"인지 "입력 오류"인지 로그 메시지를 분리해 남깁니다.
-
로그/예외 처리 규칙
- 컨텍스트 이름(작업명, 리소스 ID)을 반드시 로그 키로 포함해 추적성을 높입니다.
- 예외는 컨텍스트 내부에서 의미를 보강해 로깅하고, 상위 정책을 위해 재전파합니다.
-
재사용 함수/구조화 팁
- 반복되는 전처리-후처리(타이머, 임시설정, 트랜잭션, 권한 임시 상승)는 함수형 컨텍스트로 묶어 재사용합니다.
- 라이브러리 API로 노출할 때는
@contextmanager를 우선 고려하고, 상태가 많은 경우 클래스형 컨텍스트(__enter__,__exit__)로 전환합니다.
-
성능/메모리 체크 포인트
- 컨텍스트 매니저 자체 오버헤드는 작지만, 내부 로깅/복사(
dict.copy)가 큰 경우 누적 비용을 측정해야 합니다. ExitStack으로 다수 자원을 열 때는 가능한 한 빨리 처리하고 블록 범위를 최소화해 점유 시간을 줄입니다.
- 컨텍스트 매니저 자체 오버헤드는 작지만, 내부 로깅/복사(
오늘의 결론
한 줄 요약: contextlib는 정리 누락을 사람의 기억이 아니라 코드 구조로 막아 주는 안전장치다.
기억할 것:
try/finally를 반복해서 쓰는 순간, 컨텍스트 매니저 추출 후보입니다.- 예외를 기록해도 재전파할지 여부는 명확한 정책으로 결정해야 합니다.
- 자원 개수가 런타임에 달라지면
ExitStack이 정답에 가깝습니다.
연습문제
@contextmanager를 사용해temporary_env(key, value)를 작성해 보세요. 블록 안에서는os.environ[key]가 바뀌고, 블록을 빠져나오면 원래 값으로 복구되어야 합니다.- 파일 경로 리스트를 받아 모든 첫 줄을 읽어 오는 함수를
ExitStack으로 구현해 보세요. 중간에 존재하지 않는 파일이 있어도 이미 열린 파일이 안전하게 닫히는지 확인해 보세요. error_boundary(task_name)컨텍스트를 확장해, 예외 발생 시 재시도 횟수를 로그로 남기되 최종 실패는 반드시 상위로 올리는 버전을 만들어 보세요.
이전 강의 정답
Storage프로토콜 +MemoryStorage구현
>>> from typing import Protocol
>>>
>>> class Storage(Protocol):
... def save(self, key: str, value: str) -> None:
... ...
... def load(self, key: str) -> str | None:
... ...
...
>>> class MemoryStorage:
... def __init__(self):
... self._data: dict[str, str] = {}
... def save(self, key: str, value: str) -> None:
... self._data[key] = value
... def load(self, key: str) -> str | None:
... return self._data.get(key)
...
>>> s = MemoryStorage()
>>> s.save("lang", "python")
>>> s.load("lang")
'python'
>>> s.load("missing") is None
True
ImageResizer프로토콜 + Fake/Local 구현체 주입
>>> from typing import Protocol
>>>
>>> class ImageResizer(Protocol):
... def resize(self, path: str, width: int) -> str:
... ...
...
>>> class FakeResizer:
... def resize(self, path: str, width: int) -> str:
... return f"fake://{path}?w={width}"
...
>>> class LocalResizer:
... def resize(self, path: str, width: int) -> str:
... return f"{path.rsplit('.', 1)[0]}_{width}.jpg"
...
>>> def make_thumbnail(resizer: ImageResizer, path: str) -> str:
... return resizer.resize(path, 320)
...
>>> make_thumbnail(FakeResizer(), "cat.png")
'fake://cat.png?w=320'
>>> make_thumbnail(LocalResizer(), "cat.png")
'cat_320.jpg'
@runtime_checkable유무에 따른isinstance비교
>>> from typing import Protocol, runtime_checkable
>>>
>>> class RunnerA(Protocol):
... def run(self) -> None:
... ...
...
>>> @runtime_checkable
... class RunnerB(Protocol):
... def run(self) -> None:
... ...
...
>>> class Job:
... def run(self) -> None:
... pass
...
>>> isinstance(Job(), RunnerA)
Traceback (most recent call last):
...
TypeError: Instance and class checks can only be used with @runtime_checkable protocols
>>> isinstance(Job(), RunnerB)
True
실습 환경/재현 정보
- 실행 환경:
condaenvpython100(Python 3.11.14) - 가정한 OS: macOS/Linux 공통
- 사용 모듈:
contextlib(contextmanager,ExitStack),pathlib - 재현 체크:
- 예외가 발생해도 종료 로그/정리 코드가 항상 실행되는지
ExitStack에서 가변 자원 개수 처리 시 누수 없이 종료되는지- 블록 종료 후 임시 설정이 원복되는지