Что нужно для реализации контекстного менеджера?

Контекстный менеджер в Python: with statement и управление ресурсами

Контекстный менеджер — это объект, который управляет входом и выходом из блока кода, обычно используется для управления ресурсами (файлы, соединения, блокировки). В Python это реализуется через протокол __enter__ и __exit__.

Основы: Протокол контекстного менеджера

Для создания контекстного менеджера нужно реализовать два метода:

class MyContextManager:
    def __enter__(self):
        # Выполняется когда входим в блок with
        print("Entering context")
        # Обычно здесь инициализируем ресурсы
        return self  # Или какой-то объект
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Выполняется когда выходим из блока with
        # Даже если была ошибка!
        print("Exiting context")
        # Параметры:
        # - exc_type: тип исключения (None если не было)
        # - exc_val: значение исключения
        # - exc_tb: traceback исключения
        
        # Если вернуть True → исключение подавляется
        # Если False или ничего → исключение пробрасывается дальше
        return False

# Использование
with MyContextManager() as cm:
    print("Inside context")
    # Автоматически вызовется __exit__ даже при ошибке

# Вывод:
# Entering context
# Inside context
# Exiting context

Практические примеры

1. Управление файлом (самый частый случай)

# Файл реализует контекстный менеджер:
with open("data.txt", "r") as f:
    content = f.read()
    # Файл автоматически закроется, даже если была ошибка

# Эквивалентно:
try:
    f = open("data.txt", "r")
    content = f.read()
finally:
    f.close()  # Гарантировано вызовется

# Собственная реализация:
class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
        return False

# Использование:
with FileManager("data.txt", "r") as f:
    content = f.read()

2. Управление подключением к БД

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.conn = None
    
    def __enter__(self):
        # Подключаемся при входе
        self.conn = sqlite3.connect(self.connection_string)
        return self.conn
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Закрываемся при выходе
        if self.conn:
            self.conn.close()
        return False

# Использование:
with DatabaseConnection("db.sqlite") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()
# Соединение автоматически закроется

3. Управление блокировкой (thread lock)

import threading

lock = threading.Lock()

# Lock имеет __enter__ и __exit__
with lock:
    # Критичный раздел кода
    shared_resource.value = 42
# Lock автоматически отпустится

# Собственная реализация:
class Lock:
    def __init__(self):
        self._locked = False
    
    def __enter__(self):
        # Попробовать захватить блокировку
        while self._locked:
            pass  # Ждём
        self._locked = True
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Освободить блокировку
        self._locked = False
        return False

4. Управление временем и логированием

import time

class Timer:
    def __init__(self, name):
        self.name = name
        self.start = None
    
    def __enter__(self):
        self.start = time.time()
        print(f"Starting {self.name}")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        elapsed = time.time() - self.start
        print(f"Finished {self.name} in {elapsed:.2f}s")
        return False

# Использование:
with Timer("database query"):
    result = db.execute("SELECT * FROM large_table")
    # Выведет время выполнения

5. Обработка исключений в контекстном менеджере

class ExceptionHandler:
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            # Произошла ошибка
            print(f"Error: {exc_type.__name__}: {exc_val}")
            # Вернуть True → подавить исключение
            # Вернуть False → пробросить исключение
            return True  # Подавляем ошибку
        return False

# Использование:
with ExceptionHandler():
    x = 1 / 0  # ZeroDivisionError
    # Error напечатается, но код продолжит выполняться

print("Code continued after error")  # Это выполнится

contextlib.contextmanager — декоратор

Вместо класса с enter и exit, можно использовать декоратор:

from contextlib import contextmanager

@contextmanager
def simple_context():
    print("Entering")
    yield  # Код перед yield выполняется при __enter__
    # Код после yield выполняется при __exit__
    print("Exiting")

with simple_context():
    print("Inside")

# Более сложный пример:
@contextmanager
def database_transaction(connection):
    cursor = connection.cursor()
    try:
        yield cursor  # Передаём курсор в блок with
        connection.commit()  # Успех → коммитим
    except Exception as e:
        connection.rollback()  # Ошибка → откатываем
        raise
    finally:
        cursor.close()

with database_transaction(conn) as cursor:
    cursor.execute("INSERT INTO users VALUES (...)")

Множественные контекстные менеджеры

# Python 3.10+ синтаксис (или несколько as)
with open("input.txt") as fin, open("output.txt", "w") as fout:
    for line in fin:
        fout.write(line.upper())

# Или старший синтаксис:
with open("input.txt") as fin:
    with open("output.txt", "w") as fout:
        for line in fin:
            fout.write(line.upper())

# Или с использованием ExitStack:
from contextlib import ExitStack

with ExitStack() as stack:
    fin = stack.enter_context(open("input.txt"))
    fout = stack.enter_context(open("output.txt", "w"))
    for line in fin:
        fout.write(line.upper())

Параметры exit

class DetailedContext:
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            print("No error occurred")
        elif exc_type is ValueError:
            print(f"ValueError: {exc_val}")
            return True  # Подавляем ValueError
        elif exc_type is KeyError:
            print(f"KeyError: {exc_val}")
            return False  # Не подавляем KeyError
        else:
            # Другие ошибки
            return False

# exc_tb содержит traceback
# Можно использовать для логирования полного стека:
print(exc_tb.tb_frame)  # Информация о фрейме где произошла ошибка

Лучшие практики

# ✅ Хорошо: используй контекстные менеджеры для управления ресурсами
with open("file.txt") as f:
    data = f.read()

# ❌ Плохо: открыть, забыть закрыть
f = open("file.txt")
data = f.read()
# Файл останется открытым!

# ✅ Хорошо: явное управление ошибками
with lock:
    critical_section()

# ❌ Плохо: попробуй/отпусти вручную
lock.acquire()
try:
    critical_section()
finally:
    lock.release()

Заключение

Контекстный менеджер — это мощный инструмент для управления ресурсами в Python. Реализуется через методы __enter__ и __exit__, обеспечивает гарантированное освобождение ресурсов даже при ошибках. Используется для файлов, соединений БД, блокировок и других ресурсов. В современном Python предпочитай контекстные менеджеры явным управлению ресурсами.