JH Blog

1. 컨텍스트 관리자 기본 개념

핵심 기능

리소스 관리 자동화

파일, DB 연결, 스레드 등에 사용
with 블록 종료 시 자동 정리

예외 안전성

try-finally 블록 대체
코드 간소화

전통적 방식 vs 컨텍스트 관리자

python

# ❌ 전통적 방식 (번거로움)
file = open('data.txt', 'r')
try:
    content = file.read()
    # 작업 수행
finally:
    file.close()

# ✅ 컨텍스트 관리자 (간결)
with open('data.txt', 'r') as file:
    content = file.read()
    # 작업 수행
# 자동으로 파일 닫힘

2. 클래스 기반 구현

기본 구조

python

class ContextManager:
    def __enter__(self):
        """리소스 획득"""
        # 초기화 작업
        return self  # 또는 반환할 객체
    
    def __exit__(self, exc_type, exc_value, traceback):
        """리소스 정리"""
        # 정리 작업
        # True 반환 시 예외 억제, False/None 반환 시 예외 전파
        return False

실전 예제: 데이터베이스 연결

python

class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name
        self.connection = None
        self.cursor = None
    
    def __enter__(self):
        """연결 시작"""
        print(f"데이터베이스 연결: {self.db_name}")
        self.connection = sqlite3.connect(self.db_name)
        self.cursor = self.connection.cursor()
        return self.cursor
    
    def __exit__(self, exc_type, exc_value, traceback):
        """연결 종료"""
        if exc_type is not None:
            # 예외 발생 시 롤백
            print(f"오류 발생: {exc_value}")
            self.connection.rollback()
        else:
            # 정상 종료 시 커밋
            self.connection.commit()
        
        self.cursor.close()
        self.connection.close()
        print("데이터베이스 연결 종료")
        
        return False  # 예외를 다시 발생시킴

# 사용 예시
with DatabaseConnection('test.db') as cursor:
    cursor.execute('SELECT * FROM users')
    results = cursor.fetchall()
# 자동으로 커밋 및 연결 종료

예제: 파일 락 관리

python

import fcntl

class FileLock:
    def __init__(self, filename):
        self.filename = filename
        self.file = None
    
    def __enter__(self):
        """파일 락 획득"""
        self.file = open(self.filename, 'a')
        fcntl.flock(self.file, fcntl.LOCK_EX)
        print(f"파일 락 획득: {self.filename}")
        return self.file
    
    def __exit__(self, exc_type, exc_value, traceback):
        """파일 락 해제"""
        if self.file:
            fcntl.flock(self.file, fcntl.LOCK_UN)
            self.file.close()
            print(f"파일 락 해제: {self.filename}")
        return False

# 사용 예시
with FileLock('shared_resource.txt') as f:
    f.write('안전한 작업\n')

3. @contextmanager 데코레이터

기본 개념

특징:

제너레이터 함수로 컨텍스트 관리자 생성
클래스보다 간결
contextlib 모듈 필요

구조:

yield 전: __enter__ 역할 (리소스 할당)
yield 후: __exit__ 역할 (정리 코드)

기본 사용법

python

from contextlib import contextmanager

@contextmanager
def my_context():
    # __enter__ 역할
    print("리소스 할당")
    resource = "리소스 객체"
    
    try:
        yield resource  # 사용자에게 전달
    finally:
        # __exit__ 역할
        print("리소스 정리")

# 사용
with my_context() as res:
    print(f"사용 중: {res}")

4. 실용 예제

예제 1: 실행 시간 측정

python

import time
from contextlib import contextmanager

@contextmanager
def timer(name="작업"):
    """실행 시간 측정"""
    start = time.time()
    print(f"{name} 시작...")
    
    try:
        yield
    finally:
        elapsed = time.time() - start
        print(f"{name} 완료: {elapsed:.2f}초")

# 사용 예시
with timer("데이터 처리"):
    time.sleep(2)
    result = sum(range(1000000))

예제 2: 임시 파일 관리

python

import os
import tempfile
from contextlib import contextmanager

@contextmanager
def temporary_file(suffix='.txt'):
    """임시 파일 자동 생성 및 삭제"""
    # 임시 파일 생성
    fd, path = tempfile.mkstemp(suffix=suffix)
    os.close(fd)  # 파일 디스크립터 닫기
    
    try:
        yield path  # 파일 경로 전달
    finally:
        # 파일 삭제
        if os.path.exists(path):
            os.remove(path)
            print(f"임시 파일 삭제: {path}")

# 사용 예시
with temporary_file('.log') as temp_path:
    with open(temp_path, 'w') as f:
        f.write("임시 데이터")
    print(f"임시 파일 위치: {temp_path}")
# 자동으로 파일 삭제됨

예제 3: 데이터베이스 트랜잭션

python

import sqlite3
from contextlib import contextmanager

@contextmanager
def db_transaction(db_name):
    """데이터베이스 트랜잭션 관리"""
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    
    try:
        yield cursor  # 커서 전달
        conn.commit()  # 성공 시 커밋
        print("트랜잭션 커밋")
    except Exception as e:
        conn.rollback()  # 실패 시 롤백
        print(f"트랜잭션 롤백: {e}")
        raise
    finally:
        cursor.close()
        conn.close()
        print("연결 종료")

# 사용 예시
with db_transaction('test.db') as cursor:
    cursor.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT)')
    cursor.execute('INSERT INTO users VALUES (1, "Alice")')

예제 4: 디렉토리 변경

python

import os
from contextlib import contextmanager

@contextmanager
def change_directory(path):
    """임시로 디렉토리 변경"""
    original_dir = os.getcwd()
    
    try:
        os.chdir(path)
        yield path
    finally:
        os.chdir(original_dir)

# 사용 예시
print(f"현재 디렉토리: {os.getcwd()}")

with change_directory('/tmp'):
    print(f"변경된 디렉토리: {os.getcwd()}")
    # 작업 수행

print(f"복원된 디렉토리: {os.getcwd()}")

예제 5: 환경 변수 관리

python

import os
from contextlib import contextmanager

@contextmanager
def environment(**kwargs):
    """임시 환경 변수 설정"""
    original = {}
    
    # 기존 값 저장 및 새 값 설정
    for key, value in kwargs.items():
        original[key] = os.environ.get(key)
        os.environ[key] = str(value)
    
    try:
        yield
    finally:
        # 원래 값 복원
        for key, value in original.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value

# 사용 예시
with environment(DEBUG='true', LOG_LEVEL='info'):
    print(os.environ['DEBUG'])  # 'true'
    print(os.environ['LOG_LEVEL'])  # 'info'

# 원래 값으로 복원됨

예제 6: 표준 출력 리디렉션

python

import sys
from io import StringIO
from contextlib import contextmanager

@contextmanager
def redirect_stdout():
    """표준 출력을 캡처"""
    old_stdout = sys.stdout
    sys.stdout = captured_output = StringIO()
    
    try:
        yield captured_output
    finally:
        sys.stdout = old_stdout

# 사용 예시
with redirect_stdout() as output:
    print("이 출력은 캡처됩니다")
    print("화면에 표시되지 않습니다")

captured = output.getvalue()
print(f"캡처된 내용:\n{captured}")

5. 클래스 vs @contextmanager 비교

기준	@contextmanager	클래스
코드량	적음 (간결)	많음 (상세)
복잡도	단순 로직에 적합	복잡한 관리 가능
예외 처리	명시적 try-finally 필요	__exit__에서 통합
재사용성	함수 단위 재사용	클래스 인스턴스 재사용
상태 관리	제한적	인스턴스 변수로 자유로움
가독성	높음 (단순한 경우)	높음 (복잡한 경우)

선택 기준

python

# ✅ @contextmanager 사용 적합
# - 단순한 리소스 관리
# - 일회성 작업
# - 빠른 프로토타이핑

@contextmanager
def simple_lock():
    lock.acquire()
    try:
        yield
    finally:
        lock.release()

# ✅ 클래스 사용 적합
# - 복잡한 상태 관리
# - 여러 메서드 필요
# - 인스턴스 재사용

class ComplexResourceManager:
    def __init__(self, config):
        self.config = config
        self.resources = []
    
    def __enter__(self):
        self.setup()
        return self
    
    def __exit__(self, *exc):
        self.cleanup()
    
    def setup(self):
        # 복잡한 초기화
        pass
    
    def cleanup(self):
        # 복잡한 정리
        pass

6. 사용 시 주의사항

주의 1: yield는 한 번만

python

# ❌ 잘못된 예
@contextmanager
def bad_context():
    yield "첫 번째"
    yield "두 번째"  # 오류!

# ✅ 올바른 예
@contextmanager
def good_context():
    yield "단일 값"

주의 2: 리소스 정리는 finally 블록에서

python

# ❌ 잘못된 예
@contextmanager
def bad_cleanup():
    resource = acquire()
    yield resource
    release(resource)  # 예외 발생 시 실행 안 됨!

# ✅ 올바른 예
@contextmanager
def good_cleanup():
    resource = acquire()
    try:
        yield resource
    finally:
        release(resource)  # 항상 실행됨

주의 3: 예외 처리

python

@contextmanager
def with_exception_handling():
    try:
        yield
    except ValueError as e:
        print(f"ValueError 처리: {e}")
        # 예외를 억제하려면 아무것도 안 함
    except Exception as e:
        print(f"기타 예외: {e}")
        raise  # 예외를 다시 발생시킴

주의 4: 값 전달

python

@contextmanager
def with_value():
    value = "전달할 값"
    yield value  # 이 값이 as 절로 전달됨

# 사용
with with_value() as v:
    print(v)  # "전달할 값"

7. 고급 활용

고급 1: 중첩 컨텍스트 관리자

python

from contextlib import contextmanager, ExitStack

@contextmanager
def open_files(filenames):
    """여러 파일을 동시에 관리"""
    with ExitStack() as stack:
        files = [stack.enter_context(open(f)) for f in filenames]
        yield files

# 사용 예시
with open_files(['file1.txt', 'file2.txt', 'file3.txt']) as files:
    for f in files:
        print(f.read())

고급 2: 분산 락 시스템

python

import redis
from contextlib import contextmanager

@contextmanager
def distributed_lock(redis_client, lock_name, timeout=10):
    """Redis를 이용한 분산 락"""
    lock_key = f"lock:{lock_name}"
    lock_acquired = False
    
    try:
        # 락 획득 시도
        lock_acquired = redis_client.set(
            lock_key, 
            "1", 
            nx=True,  # 키가 없을 때만 설정
            ex=timeout  # 자동 만료
        )
        
        if not lock_acquired:
            raise RuntimeError(f"락 획득 실패: {lock_name}")
        
        yield
    finally:
        # 락 해제
        if lock_acquired:
            redis_client.delete(lock_key)

# 사용 예시
r = redis.Redis()
with distributed_lock(r, "my_resource"):
    # 크리티컬 섹션
    print("배타적 작업 수행")

고급 3: 동적 환경 설정

python

from contextlib import contextmanager

class Config:
    DEBUG = False
    LOG_LEVEL = "info"

@contextmanager
def temporary_config(**kwargs):
    """임시 설정 적용"""
    # 기존 설정 백업
    original = {}
    for key, value in kwargs.items():
        if hasattr(Config, key):
            original[key] = getattr(Config, key)
        setattr(Config, key, value)
    
    try:
        yield Config
    finally:
        # 설정 복원
        for key, value in original.items():
            setattr(Config, key, value)

# 사용 예시
print(Config.DEBUG)  # False

with temporary_config(DEBUG=True, LOG_LEVEL="debug"):
    print(Config.DEBUG)  # True
    print(Config.LOG_LEVEL)  # "debug"

print(Config.DEBUG)  # False (복원됨)

고급 4: 커스텀 예외 처리

python

from contextlib import contextmanager

@contextmanager
def handle_specific_errors(*error_types):
    """특정 예외만 처리"""
    try:
        yield
    except error_types as e:
        print(f"예상된 오류 처리: {type(e).__name__}: {e}")
        # 예외를 억제 (다시 발생시키지 않음)
    except Exception as e:
        print(f"예상치 못한 오류: {type(e).__name__}: {e}")
        raise  # 다시 발생시킴

# 사용 예시
with handle_specific_errors(ValueError, TypeError):
    int("invalid")  # ValueError - 처리됨

# KeyError는 처리되지 않아 프로그램 종료

고급 5: 재진입 가능한 컨텍스트 관리자

python

from threading import RLock
from contextlib import contextmanager

class ReentrantContext:
    def __init__(self):
        self.lock = RLock()
        self.count = 0
    
    def __enter__(self):
        self.lock.acquire()
        self.count += 1
        print(f"진입 (깊이: {self.count})")
        return self
    
    def __exit__(self, *exc):
        self.count -= 1
        print(f"탈출 (깊이: {self.count})")
        self.lock.release()

# 사용 예시
ctx = ReentrantContext()

with ctx:
    print("첫 번째 레벨")
    with ctx:
        print("두 번째 레벨 (같은 객체)")

8. 실전 패턴

패턴 1: 리소스 풀 관리

python

from contextlib import contextmanager
from queue import Queue

class ConnectionPool:
    def __init__(self, size):
        self.pool = Queue(maxsize=size)
        for _ in range(size):
            self.pool.put(self.create_connection())
    
    def create_connection(self):
        return {"id": id(object())}
    
    @contextmanager
    def get_connection(self):
        conn = self.pool.get()
        try:
            yield conn
        finally:
            self.pool.put(conn)

# 사용
pool = ConnectionPool(5)

with pool.get_connection() as conn:
    print(f"연결 사용: {conn['id']}")

패턴 2: 로깅 컨텍스트

python

import logging
from contextlib import contextmanager

@contextmanager
def log_context(name, level=logging.INFO):
    """로그 레벨 임시 변경"""
    logger = logging.getLogger(name)
    original_level = logger.level
    
    logger.setLevel(level)
    logger.info(f"로그 레벨 변경: {logging.getLevelName(level)}")
    
    try:
        yield logger
    finally:
        logger.setLevel(original_level)
        logger.info("로그 레벨 복원")

# 사용
with log_context("myapp", logging.DEBUG) as logger:
    logger.debug("디버그 메시지")

패턴 3: 트랜잭션 재시도

python

from contextlib import contextmanager
import time

@contextmanager
def retry_transaction(max_retries=3, delay=1):
    """트랜잭션 재시도"""
    for attempt in range(max_retries):
        try:
            yield attempt
            break  # 성공 시 종료
        except Exception as e:
            if attempt < max_retries - 1:
                print(f"재시도 {attempt + 1}/{max_retries}: {e}")
                time.sleep(delay)
            else:
                print("최대 재시도 횟수 초과")
                raise

# 사용
with retry_transaction(max_retries=3) as attempt:
    print(f"시도 {attempt + 1}")
    # 트랜잭션 로직

9. 종합 정리

핵심 개념

개념	설명
enter	리소스 획득, 초기화
exit	리소스 정리, 예외 처리
@contextmanager	제너레이터 기반 간결한 구현
yield	리소스 전달 지점
try-finally	안전한 정리 보장

사용 체크리스트

✅ 구현 시:

리소스 할당 코드 작성
try-finally로 정리 보장
예외 처리 로직 추가
값 전달 여부 결정

✅ 선택 기준:

단순한 경우: @contextmanager
복잡한 경우: 클래스 기반
재사용 필요: 클래스 기반
일회성 작업: @contextmanager

일반적인 사용 사례

파일 관리: 자동 닫기
DB 연결: 트랜잭션 관리
락/세마포어: 동기화
임시 상태: 환경 변수, 설정
성능 측정: 시간, 메모리
예외 처리: 특정 오류 억제