본문으로 건너뛰기
Advertisement

컨텍스트 매니저 — with 문, __enter__/__exit__, contextlib

with 문은 리소스를 안전하게 획득하고 해제하는 Python의 핵심 패턴입니다. 파일, 데이터베이스 연결, 락(lock), 임시 디렉터리 등 종료 처리가 필요한 모든 곳에 활용됩니다.

with 문 동작 원리​

# with 문의 기본 구조
with expression as variable:
    body

# 위 코드는 아래와 동일하게 동작합니다
manager = expression
variable = manager.__enter__()
try:
    body
except:
    if not manager.__exit__(*sys.exc_info()):
        raise
else:
    manager.__exit__(None, None, None)

핵심 규칙

  • __enter__는 리소스를 획득하고, as 변수에 바인딩할 값을 반환합니다.
  • __exit__는 블록 종료 시 항상 호출됩니다 (예외 발생 여부 무관).
  • __exit__가 True를 반환하면 예외를 억제합니다.

__enter__ / __exit__ 프로토콜 직접 구현​

class ManagedFile:
    """파일 컨텍스트 매니저"""

    def __init__(self, path: str, mode: str = "r", encoding: str = "utf-8") -> None:
        self.path = path
        self.mode = mode
        self.encoding = encoding
        self.file = None

    def __enter__(self):
        self.file = open(self.path, self.mode, encoding=self.encoding)
        return self.file   # as 변수에 바인딩

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        """
        exc_type:  예외 클래스 (없으면 None)
        exc_val:   예외 인스턴스 (없으면 None)
        exc_tb:    트레이스백 객체 (없으면 None)
        반환값:    True → 예외 억제, False/None → 예외 전파
        """
        if self.file:
            self.file.close()
        # False 반환: 예외를 억제하지 않고 그대로 전파
        return False

# 사용
with ManagedFile("data.txt") as f:
    content = f.read()

__exit__ 매개변수 활용​

class SafeWrite:
    """예외 발생 시 파일을 자동으로 삭제하는 컨텍스트 매니저"""

    def __init__(self, path: str) -> None:
        self.path = path
        self.file = None

    def __enter__(self):
        self.file = open(self.path, "w", encoding="utf-8")
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        self.file.close()
        if exc_type is not None:
            # 예외 발생 시 불완전한 파일 삭제
            import os
            os.unlink(self.path)
            print(f"오류로 인해 파일 삭제: {self.path}")
        return False  # 예외 전파

contextlib.contextmanager 데코레이터​

contextlib.contextmanager를 사용하면 제너레이터 함수로 컨텍스트 매니저를 간결하게 작성할 수 있습니다.

from contextlib import contextmanager

@contextmanager
def managed_file(path: str, mode: str = "r"):
    file = open(path, mode, encoding="utf-8")
    try:
        yield file          # ← yield 이전: __enter__
    finally:
        file.close()        # ← yield 이후: __exit__

# 사용
with managed_file("data.txt") as f:
    print(f.read())

yield 앞은 __enter__, yield 뒤는 __exit__에 해당합니다. yield로 반환한 값이 as 변수에 바인딩됩니다.

예외 처리 포함 패턴​

from contextlib import contextmanager

@contextmanager
def transaction(conn):
    """데이터베이스 트랜잭션 컨텍스트 매니저"""
    try:
        yield conn
        conn.commit()           # 성공 시 커밋
    except Exception:
        conn.rollback()         # 예외 시 롤백
        raise                   # 예외 재발생
    finally:
        conn.close()            # 항상 연결 해제

contextlib.suppress — 예외 무시​

특정 예외를 조용히 무시하고 싶을 때 사용합니다.

from contextlib import suppress
import os

# suppress 없이
try:
    os.remove("temp.txt")
except FileNotFoundError:
    pass

# suppress 사용 — 더 간결
with suppress(FileNotFoundError):
    os.remove("temp.txt")

# 여러 예외 억제
with suppress(FileNotFoundError, PermissionError):
    os.remove("protected.txt")

주의: suppress는 예외를 완전히 무시하므로, 오류를 알아야 하는 상황에서는 사용하지 마십시오.

contextlib.ExitStack — 동적 컨텍스트 스택​

파일 개수가 동적으로 결정되거나, 조건부로 컨텍스트 매니저를 활성화해야 할 때 사용합니다.

from contextlib import ExitStack

# 여러 파일 동시 열기 (수가 동적)
def merge_files(input_paths: list[str], output_path: str) -> None:
    with ExitStack() as stack:
        # 동적으로 파일들을 스택에 추가
        files = [
            stack.enter_context(open(p, "r", encoding="utf-8"))
            for p in input_paths
        ]
        out = stack.enter_context(open(output_path, "w", encoding="utf-8"))

        for f in files:
            out.write(f.read())

# 조건부 컨텍스트
def process(data, use_lock: bool = False):
    with ExitStack() as stack:
        if use_lock:
            import threading
            lock = threading.Lock()
            stack.enter_context(lock)
        # 이후 코드는 use_lock 여부와 무관하게 동일
        do_work(data)

ExitStack으로 콜백 등록​

from contextlib import ExitStack

with ExitStack() as stack:
    # 종료 시 호출될 콜백 등록
    stack.callback(print, "정리 1 완료")
    stack.callback(print, "정리 2 완료")
    print("작업 수행")

# 출력:
# 작업 수행
# 정리 2 완료  ← LIFO 순서
# 정리 1 완료

실전 예제 1 — DB 트랜잭션​

import sqlite3
from contextlib import contextmanager
from typing import Generator

@contextmanager
def db_transaction(db_path: str) -> Generator[sqlite3.Connection, None, None]:
    """
    SQLite 트랜잭션을 자동으로 관리합니다.
    예외 발생 시 롤백, 정상 종료 시 커밋합니다.
    """
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    except sqlite3.Error as e:
        conn.rollback()
        raise RuntimeError(f"DB 트랜잭션 실패: {e}") from e
    finally:
        conn.close()

# 사용
def transfer_points(db: str, from_id: int, to_id: int, amount: int):
    with db_transaction(db) as conn:
        conn.execute(
            "UPDATE users SET points = points - ? WHERE id = ?",
            (amount, from_id)
        )
        conn.execute(
            "UPDATE users SET points = points + ? WHERE id = ?",
            (amount, to_id)
        )
        # 예외 발생 시 두 쿼리 모두 롤백

실전 예제 2 — 타이머​

import time
from contextlib import contextmanager
from dataclasses import dataclass, field

@dataclass
class TimerResult:
    elapsed: float = 0.0
    label: str = ""

    def __str__(self) -> str:
        return f"[{self.label}] {self.elapsed:.4f}초"

@contextmanager
def timer(label: str = ""):
    result = TimerResult(label=label)
    start = time.perf_counter()
    try:
        yield result
    finally:
        result.elapsed = time.perf_counter() - start
        print(result)

# 사용
with timer("데이터 처리") as t:
    # 처리할 작업
    data = list(range(1_000_000))
    total = sum(data)

# [데이터 처리] 0.0312초

실전 예제 3 — 임시 디렉터리​

import shutil
import tempfile
from contextlib import contextmanager
from pathlib import Path

@contextmanager
def temp_directory(prefix: str = "tmp_"):
    """임시 디렉터리를 생성하고 종료 시 자동으로 삭제합니다."""
    tmp_dir = Path(tempfile.mkdtemp(prefix=prefix))
    try:
        yield tmp_dir
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)

# 사용
with temp_directory("build_") as tmp:
    (tmp / "output.txt").write_text("임시 파일")
    (tmp / "subdir").mkdir()
    # with 블록 종료 시 tmp 디렉터리 전체 삭제

# 표준 라이브러리: tempfile.TemporaryDirectory()
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
    path = Path(tmpdir) / "file.txt"
    path.write_text("내용")

비동기 컨텍스트 매니저 (async with)​

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def async_timer(label: str = ""):
    start = asyncio.get_event_loop().time()
    try:
        yield
    finally:
        elapsed = asyncio.get_event_loop().time() - start
        print(f"[{label}] {elapsed:.4f}초")

async def main():
    async with async_timer("비동기 작업"):
        await asyncio.sleep(0.1)

asyncio.run(main())

고수 팁​

팁 1 — __enter__에서 self를 반환하는 패턴

class Connection:
    def __enter__(self):
        self.connect()
        return self   # self를 반환하면 with ... as conn: conn.method() 가능

    def __exit__(self, *args):
        self.close()
        return False

팁 2 — contextlib.contextmanager와 async 함께 사용

asynccontextmanager는 contextmanager의 비동기 버전입니다. async for, async with를 사용하는 코드에 필수입니다.

팁 3 — 중첩 with 대신 단일 with 사용 (Python 3.10+)

# 구식
with open("a.txt") as f1:
    with open("b.txt") as f2:
        pass

# Python 3.10+ 괄호 문법
with (
    open("a.txt") as f1,
    open("b.txt") as f2,
):
    pass

팁 4 — nullcontext — 조건부 컨텍스트 매니저

from contextlib import nullcontext

def process(data, lock=None):
    # lock이 None이면 아무것도 하지 않는 컨텍스트 사용
    with lock if lock is not None else nullcontext():
        do_work(data)
Advertisement
Advertisement