Что такое контекстные менеджеры?

Контекстные менеджеры в Python - это объекты, которые определяют среду выполнения для блока кода с помощью оператора with. Они обеспечивают точное выделение и освобождение ресурсов, гарантируя выполнение определенных операций до и после выполнения блока кода. Контекстные менеджеры особенно полезны при работе с файлами, сетевыми соединениями, блокировками и другими ресурсами, которые требуют корректного закрытия или освобождения.

Основные преимущества контекстных менеджеров

Оператор with

Оператор with - это основной способ использования контекстных менеджеров в Python. Он обеспечивает автоматическое управление ресурсами и обработку исключений:

Базовое использование with 
# Традиционный способ открытия файла
file = open("example.txt", "r")
try:
    content = file.read()
    print(content)
finally:
    file.close()  # Файл должен быть закрыт вручную

# С использованием контекстного менеджера
with open("example.txt", "r") as file:
    content = file.read()
    print(content)
# Файл автоматически закрывается при выходе из блока with

# Работа с несколькими файлами
with open("input.txt", "r") as input_file, \
     open("output.txt", "w") as output_file:
    data = input_file.read()
    output_file.write(data.upper())

Создание контекстных менеджеров с помощью классов

Для создания собственного контекстного менеджера с помощью класса необходимо реализовать два метода: __enter__ и __exit__:

Контекстный менеджер на основе класса 
class Timer:
    def __init__(self):
        self.start = None
        self.end = None
    
    def __enter__(self):
        import time
        self.start = time.time()
        print("Таймер запущен")
        return self  # Возвращаем объект для использования в блоке with
    
    def __exit__(self, exc_type, exc_value, traceback):
        import time
        self.end = time.time()
        print(f"Таймер остановлен. Прошло времени: {self.end - self.start:.4f} секунд")
        # Возвращаем False, чтобы исключения не подавлялись
        return False

# Использование контекстного менеджера
with Timer() as timer:
    # Имитируем выполнение какой-то работы
    import time
    time.sleep(2)
    print("Работа выполнена")

# Пример контекстного менеджера для подключения к базе данных
class DatabaseConnection:
    def __init__(self, host, port, database):
        self.host = host
        self.port = port
        self.database = database
        self.connection = None
    
    def __enter__(self):
        print(f"Подключение к базе данных {self.database} на {self.host}:{self.port}")
        # Здесь был бы код для установки реального соединения
        self.connection = "connection_object"
        return self.connection
    
    def __exit__(self, exc_type, exc_value, traceback):
        print(f"Закрытие соединения с базой данных {self.database}")
        # Здесь был бы код для закрытия реального соединения
        self.connection = None
        # Если возникло исключение, решаем, подавлять ли его
        if exc_type is not None:
            print(f"Произошла ошибка: {exc_value}")
        return False  # Не подавляем исключения

# Использование контекстного менеджера для подключения к БД
try:
    with DatabaseConnection("localhost", 5432, "my_database") as connection:
        print(f"Выполняем операции с БД: {connection}")
        # Здесь были бы реальные операции с базой данных
except Exception as e:
    print(f"Ошибка при работе с базой данных: {e}")

Создание контекстных менеджеров с помощью contextlib

Модуль contextlib предоставляет удобные инструменты для создания контекстных менеджеров, особенно декоратор @contextmanager:

Контекстные менеджеры с contextlib 
from contextlib import contextmanager
import time

# Создание контекстного менеджера с помощью декоратора
@contextmanager
def timer():
    start = time.time()
    print("Начало измерения времени")
    try:
        yield  # Передаем управление блоку with
    finally:
        end = time.time()
        print(f"Измерение завершено. Прошло: {end - start:.4f} секунд")

# Использование контекстного менеджера
with timer():
    time.sleep(1)
    print("Выполняем какую-то работу...")

# Контекстный менеджер для временного изменения переменной окружения
import os

@contextmanager
def temporary_environment_variable(name, value):
    old_value = os.environ.get(name)
    os.environ[name] = value
    try:
        yield
    finally:
        if old_value is None:
            del os.environ[name]
        else:
            os.environ[name] = old_value

# Использование контекстного менеджера для переменных окружения
print(f"PATH до изменения: {os.environ.get('PATH', 'не задан')[:50]}...")
with temporary_environment_variable("TEMP_VAR", "test_value"):
    print(f"TEMP_VAR во время блока: {os.environ.get('TEMP_VAR')}")
print(f"TEMP_VAR после блока: {os.environ.get('TEMP_VAR', 'не существует')}")

# Контекстный менеджер для подавления исключений определенного типа
@contextmanager
def suppress_exceptions(*exception_types):
    try:
        yield
    except exception_types as e:
        print(f"Подавлено исключение: {type(e).__name__}: {e}")

# Использование контекстного менеджера для подавления исключений
with suppress_exceptions(ZeroDivisionError, ValueError):
    result = 10 / 0  # ZeroDivisionError будет подавлено
    print(f"Результат: {result}")

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

Практическое применение контекстных менеджеров

Контекстные менеджеры находят применение в различных сценариях, где требуется гарантированное управление ресурсами:

Управление блокировками 
import threading
import time

# Создаем блокировку
lock = threading.Lock()

# Без контекстного менеджера
lock.acquire()
try:
    print("Критическая секция без контекстного менеджера")
    time.sleep(1)
finally:
    lock.release()

# С контекстным менеджером
with lock:
    print("Критическая секция с контекстным менеджером")
    time.sleep(1)
# Блокировка автоматически освобождается

# Собственный контекстный менеджер для блокировок с таймаутом
@contextmanager
def lock_with_timeout(lock, timeout=5):
    if lock.acquire(timeout=timeout):
        try:
            yield
        finally:
            lock.release()
    else:
        raise TimeoutError(f"Не удалось получить блокировку за {timeout} секунд")

# Использование контекстного менеджера с таймаутом
try:
    with lock_with_timeout(lock, timeout=2):
        print("Работаем с блокировкой")
except TimeoutError as e:
    print(f"Ошибка: {e}")
Управление изменениями в списках 
# Контекстный менеджер для временного изменения списка
@contextmanager
def temporary_list_change(lst, new_elements):
    original_length = len(lst)
    try:
        lst.extend(new_elements)
        yield lst
    finally:
        # Восстанавливаем оригинальное состояние списка
        del lst[original_length:]

# Использование контекстного менеджера для временного изменения списка
my_list = [1, 2, 3]
print(f"Оригинальный список: {my_list}")

with temporary_list_change(my_list, [4, 5, 6]):
    print(f"Список внутри блока: {my_list}")
    # Выполняем какие-то операции с расширенным списком

print(f"Список после блока: {my_list}")

# Контекстный менеджер для работы с кэшем
class CacheManager:
    def __init__(self):
        self.cache = {}
        self.original_cache = {}
    
    def __enter__(self):
        self.original_cache = self.cache.copy()
        return self.cache
    
    def __exit__(self, exc_type, exc_value, traceback):
        # Восстанавливаем оригинальный кэш при выходе
        self.cache.clear()
        self.cache.update(self.original_cache)
        return False

# Использование менеджера кэша
cache_manager = CacheManager()
cache_manager.cache["key1"] = "value1"

with cache_manager as cache:
    cache["key2"] = "value2"
    print(f"Кэш внутри блока: {cache}")

print(f"Кэш после блока: {cache_manager.cache}")

Продвинутые возможности контекстных менеджеров

Рассмотрим более сложные примеры использования контекстных менеджеров:

Вложенные контекстные менеджеры 
# Пример с вложенными контекстными менеджерами
@contextmanager
def logging(operation_name):
    print(f"Начало операции: {operation_name}")
    try:
        yield
    except Exception as e:
        print(f"Ошибка в операции {operation_name}: {e}")
        raise
    else:
        print(f"Успешное завершение операции: {operation_name}")

@contextmanager
def transaction():
    print("Начало транзакции")
    try:
        yield
    except Exception as e:
        print("Откат транзакции из-за ошибки")
        raise
    else:
        print("Фиксация транзакции")

# Вложенные контекстные менеджеры
try:
    with logging("Обработка данных"):
        with transaction():
            print("Выполняем обработку данных...")
            # Имитируем ошибку для демонстрации
            # raise ValueError("Ошибка обработки")
except ValueError as e:
    print(f"Обработанная ошибка: {e}")

# Использование ExitStack для динамического управления контекстными менеджерами
from contextlib import ExitStack

def process_files(file_names):
    with ExitStack() as stack:
        files = [stack.enter_context(open(name, 'w')) for name in file_names]
        for i, file in enumerate(files):
            file.write(f"Данные файла {i+1}\n")
        print("Все файлы успешно записаны")
    # Все файлы автоматически закрываются при выходе из блока

# Создаем несколько файлов
process_files(["файл1.txt", "файл2.txt", "файл3.txt"])

Упражнения

Упражнение 1: Контекстный менеджер для подключения к базе данных

Создайте контекстный менеджер для подключения к базе данных, который будет автоматически закрывать соединение при выходе из блока. Реализуйте его двумя способами: с помощью класса и с помощью декоратора @contextmanager.

# Решение с помощью класса
class ПодключениеКБД:
    def __init__(self, хост, порт, база_данных):
        self.хост = хост
        self.порт = порт
        self.база_данных = база_данных
        self.соединение = None
    
    def __enter__(self):
        print(f"Подключение к {self.база_данных} на {self.хост}:{self.порт}")
        # Здесь был бы реальный код подключения
        self.соединение = f"соединение_с_{self.база_данных}"
        return self.соединение
    
    def __exit__(self, exc_type, exc_value, traceback):
        print(f"Закрытие соединения с {self.база_данных}")
        self.соединение = None
        return False

# Решение с помощью @contextmanager
from contextlib import contextmanager

@contextmanager
def подключение_к_бд(хост, порт, база_данных):
    print(f"Подключение к {база_данных} на {хост}:{порт}")
    соединение = f"соединение_с_{база_данных}"
    try:
        yield соединение
    finally:
        print(f"Закрытие соединения с {база_данных}")

# Тестирование
print("Тестирование класса:")
with ПодключениеКБД("localhost", 5432, "тестовая_бд") as соединение:
    print(f"Работа с БД: {соединение}")

print("\nТестирование декоратора:")
with подключение_к_бд("localhost", 5432, "тестовая_бд") as соединение:
    print(f"Работа с БД: {соединение}")
Упражнение 2: Контекстный менеджер для изменения рабочей директории

Создайте контекстный менеджер, который временно изменяет рабочую директорию и возвращает исходную директорию при выходе из блока.

import os

@contextmanager
def временная_директория(новая_директория):
    оригинальная_директория = os.getcwd()
    try:
        os.chdir(новая_директория)
        yield новая_директория
    finally:
        os.chdir(оригинальная_директория)

# Тестирование
print(f"Текущая директория: {os.getcwd()}")

try:
    with временная_директория("/tmp") as temp_dir:
        print(f"Временная директория: {os.getcwd()}")
        # Выполняем какие-то операции в новой директории
except FileNotFoundError:
    print("Директория /tmp не найдена (возможно, вы используете Windows)")

print(f"Возвращены в оригинальную директорию: {os.getcwd()}")
Упражнение 3: Контекстный менеджер для подавления предупреждений

Создайте контекстный менеджер, который временно подавляет предупреждения определенного типа.

import warnings

@contextmanager
def подавление_предупреждений(*типы_предупреждений):
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", типы_предупреждений if типы_предупреждений else Warning)
        yield

# Тестирование
print("Без подавления предупреждений:")
warnings.warn("Это тестовое предупреждение", UserWarning)

print("\nС подавлением предупреждений:")
with подавление_предупреждений(UserWarning):
    warnings.warn("Это предупреждение будет подавлено", UserWarning)
    warnings.warn("Это предупреждение тоже будет подавлено", DeprecationWarning)

print("Программа продолжает работу")
Предыдущий урок Следующий урок