Какие методы необходимо определить в классе, чтобы создать контекстный менеджер с использованием оператора with?

Какие методы определить для контекстного менеджера (with)

Контекстные менеджеры — это мощный инструмент для управления ресурсами. За 10+ лет практики я использую их во всех своих проектах.

1. Основные методы: enter и exit

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

class CustomContextManager:
    """Простой контекстный менеджер"""
    
    def __enter__(self):
        """
        Вызывается при входе в блок with
        Возвращает объект, который присваивается переменной после as
        """
        print('Входим в контекст')
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Вызывается при выходе из блока with
        exc_type: тип исключения (или None если нет)
        exc_val: значение исключения (или None)
        exc_tb: traceback исключения (или None)
        
        Возвращаем True если обработали исключение
        Возвращаем False или None если хотим распространить исключение
        """
        print('Выходим из контекста')
        return False  # Не подавляем исключения

# Использование
with CustomContextManager() as ctx:
    print('Работаем в контексте')
    # ctx содержит то, что вернул __enter__

print('После контекста')

# Вывод:
# Входим в контекст
# Работаем в контексте
# Выходим из контекста
# После контекста

2. Практический пример: управление файлом

class FileManager:
    """Контекстный менеджер для работы с файлами"""
    
    def __init__(self, filename, mode='r'):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        """Открываем файл при входе"""
        print(f'Открываем файл: {self.filename}')
        self.file = open(self.filename, self.mode)
        return self.file
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Закрываем файл при выходе"""
        if self.file:
            print(f'Закрываем файл: {self.filename}')
            self.file.close()
        
        if exc_type is not None:
            print(f'Произошла ошибка: {exc_val}')
        
        return False  # Не подавляем ошибки

# Использование
with FileManager('data.txt', 'w') as f:
    f.write('Hello World')

# Файл автоматически закрывается после блока

3. Параметры exit

class ErrorHandlingManager:
    """Демонстрирует параметры __exit__"""
    
    def __enter__(self):
        print('__enter__ вызван')
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        exc_type: <class 'ValueError'> или None
        exc_val: ValueError('Invalid value')
        exc_tb: <traceback object ...>
        """
        if exc_type is None:
            print('Контекст завершен без ошибок')
        else:
            print(f'Тип ошибки: {exc_type.__name__}')
            print(f'Сообщение: {exc_val}')
        
        # Возвращаем True чтобы подавить исключение
        return isinstance(exc_type, ValueError)

# Нормальное выполнение
with ErrorHandlingManager() as manager:
    print('Нормальный код')

# С ошибкой (которая будет подавлена)
with ErrorHandlingManager() as manager:
    raise ValueError('Это будет перехвачено')

print('Программа продолжает работу')

4. Контекстный менеджер для управления БД

class DatabaseConnection:
    """Контекстный менеджер для БД подключения"""
    
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        print(f'Подключаемся к БД: {self.connection_string}')
        # Имитация подключения
        self.connection = {'status': 'connected'}
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.connection:
            print('Закрываем подключение к БД')
            self.connection = None
        
        if exc_type is not None:
            print(f'Ошибка БД: {exc_val}. Откатываем транзакцию.')
            return False  # Распространяем ошибку
        
        print('Фиксируем транзакцию')
        return False

# Использование
try:
    with DatabaseConnection('postgres://localhost/mydb') as db:
        print(f'Работаем с БД: {db}')
        # Выполняем операции
except Exception as e:
    print(f'Ошибка подключения: {e}')

5. Стек контекстных менеджеров

class ResourceA:
    def __enter__(self):
        print('Открываем ресурс A')
        return 'A'
    
    def __exit__(self, *args):
        print('Закрываем ресурс A')

class ResourceB:
    def __enter__(self):
        print('Открываем ресурс B')
        return 'B'
    
    def __exit__(self, *args):
        print('Закрываем ресурс B')

# Множественные контексты
with ResourceA() as a, ResourceB() as b:
    print(f'Работаем с {a} и {b}')

# Вывод (в том порядке они закрываются в обратном):
# Открываем ресурс A
# Открываем ресурс B
# Работаем с A и B
# Закрываем ресурс B
# Закрываем ресурс A

6. Использование @contextmanager декоратора

from contextlib import contextmanager

@contextmanager
def simple_context_manager():
    """Создаем контекстный менеджер через decorator"""
    # Код перед __enter__
    print('Вход')
    
    try:
        # yield — это точка входа в контекст
        yield 'resource'  # То, что вернет __enter__
    
    finally:
        # Код перед __exit__
        print('Выход')

# Использование
with simple_context_manager() as resource:
    print(f'Используем {resource}')

# Вывод:
# Вход
# Используем resource
# Выход

7. Контекстный менеджер с обработкой ошибок

@contextmanager
def error_handling_context():
    """Контекстный менеджер с обработкой исключений"""
    print('Инициализация')
    
    try:
        yield 'resource'
    
    except ValueError as e:
        print(f'Перехватили ValueError: {e}')
        # Если хотим подавить исключение
        # pass
        # Если хотим распространить
        raise
    
    except Exception as e:
        print(f'Неожиданная ошибка: {e}')
        raise
    
    finally:
        print('Очистка')

# Использование
try:
    with error_handling_context() as resource:
        # raise ValueError('Тестовая ошибка')
        print(f'Работаем с {resource}')
except ValueError:
    print('Ошибка распространена и обработана снаружи')

8. Примеры из стандартной библиотеки

# Управление временем
from contextlib import contextmanager
import time

@contextmanager
def timer(name):
    """Контекстный менеджер для измерения времени"""
    start = time.time()
    try:
        yield
    finally:
        elapsed = time.time() - start
        print(f'{name} занял {elapsed:.3f}s')

with timer('Операция'):
    time.sleep(0.1)

# Выход:
# Операция занял 0.100s

# Подавление исключений
from contextlib import suppress

with suppress(FileNotFoundError):
    # Исключение FileNotFoundError будет подавлено
    open('nonexistent.txt')

print('Программа продолжает работу')  # Этот код выполнится

9. Контекстный менеджер для транзакций

class Transaction:
    """Контекстный менеджер для управления транзакциями"""
    
    def __init__(self, db):
        self.db = db
        self.transaction_id = None
    
    def __enter__(self):
        print('Начинаем транзакцию')
        self.transaction_id = id(self)
        self.db.begin_transaction()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            # Нет ошибок — фиксируем
            print('Фиксируем транзакцию')
            self.db.commit()
        else:
            # Есть ошибка — откатываем
            print(f'Откатываем транзакцию: {exc_val}')
            self.db.rollback()
        
        return False  # Распространяем ошибку

class FakeDB:
    def begin_transaction(self):
        pass
    def commit(self):
        print('COMMIT')
    def rollback(self):
        print('ROLLBACK')

db = FakeDB()

try:
    with Transaction(db):
        print('Выполняем операции')
        # raise Exception('Ошибка БД')
except:
    pass

10. Advanced: вложенные контекстные менеджеры

from contextlib import ExitStack

def manage_multiple_files(filenames):
    """Управляет несколькими файлами динамически"""
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(name, 'r'))
            for name in filenames
        ]
        
        for f in files:
            print(f.read()[:50])  # Первые 50 символов
        # Все файлы автоматически закрываются

# Это эквивалентно:
# with open(f1) as f1, open(f2) as f2, open(f3) as f3:
#     ...

Сравнение способов

"""
╔════════════════════════════╦──────────────╦────────────╗
║ Метод                      ║ Сложность    ║ Применение ║
╠════════════════════════════╬──────────────╬────────────╣
║ Класс с __enter__/__exit__ ║ Высокая      ║ Сложные    ║
║ @contextmanager decorator  ║ Средняя      ║ Типовые    ║
║ from contextlib import ... ║ Низкая       ║ Простые    ║
╚════════════════════════════╩──────────────╩────────────╝
"""

Мой Чеклист при Создании Context Manager

class BestPracticesContextManager:
    """Контекстный менеджер с best practices"""
    
    def __init__(self, resource_name):
        self.resource_name = resource_name
        self.resource = None
    
    def __enter__(self):
        # 1. Инициализируем ресурс
        # 2. Логируем
        # 3. Возвращаем объект для использования
        self.resource = self._acquire_resource()
        return self.resource
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # 1. Логируем выход
        # 2. Очищаем ресурсы
        # 3. Обрабатываем ошибки если нужно
        # 4. НЕ подавляем исключения (return False)
        self._release_resource()
        return False
    
    def _acquire_resource(self):
        return f'Resource: {self.resource_name}'
    
    def _release_resource(self):
        if self.resource:
            self.resource = None

Заключение

Контекстные менеджеры необходимы для:

• Управления ресурсами (файлы, подключения, блокировки)
• Гарантии выполнения cleanup кода
• Обработки исключений
• Сохранения чистоты и читаемости кода

Главные методы:

• __enter__ — инициализация
• __exit__ — очистка, обработка ошибок

Или используй @contextmanager декоратор для простых случаев.