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

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

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

Основной синтаксис: with statement

with some_context_manager() as resource:
    # Используем ресурс
    # При выходе ресурс автоматически освободится
    pass

В этом коде:

1. Вызывается __enter__(), который инициализирует ресурс
2. Результат присваивается переменной после as
3. Выполняется блок кода
4. Вызывается __exit__() для очистки, даже если было исключение

Создание собственного контекстного менеджера

Способ 1: Через класс с методами __enter__ и __exit__

class DatabaseConnection:
    def __enter__(self):
        print("Подключение к БД")
        self.connection = connect_to_db()
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Закрытие подключения")
        self.connection.close()
        # Возвращаем True, чтобы подавить исключение
        return False

# Использование
with DatabaseConnection() as conn:
    conn.execute("SELECT * FROM users")

Способ 2: Через декоратор @contextmanager

from contextlib import contextmanager

@contextmanager
def database_connection():
    print("Подключение к БД")
    conn = connect_to_db()
    try:
        yield conn  # Результат, который получится в переменной после as
    finally:
        print("Закрытие подключения")
        conn.close()

# Использование
with database_connection() as conn:
    conn.execute("SELECT * FROM users")

Параметры __exit__

def __exit__(self, exc_type, exc_val, exc_tb):
    # exc_type - тип исключения (None если исключения не было)
    # exc_val - значение исключения
    # exc_tb - traceback исключения
    
    if exc_type is not None:
        print(f"Произошла ошибка: {exc_val}")
    
    # Возвращаем True, если обработали исключение
    # Возвращаем False, чтобы пробросить его дальше
    return False

Практические примеры из реальной работы

Управление файлами:

with open(data.csv, r) as f:
    data = f.read()
    # Файл автоматически закроется при выходе

Управление транзакциями БД:

from contextlib import contextmanager

@contextmanager
def transaction(db):
    try:
        db.begin()
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise

with transaction(db) as conn:
    conn.insert_user(name="John")
    conn.insert_order(user_id=1)

Управление блокировками (Threading):

import threading

lock = threading.Lock()

with lock:
    # Критическая секция, защищённая блокировкой
    shared_resource += 1
# Блокировка автоматически освобождается

Временная директория:

from tempfile import TemporaryDirectory

with TemporaryDirectory() as tmpdir:
    # Работаем с временной папкой
    create_file(f"{tmpdir}/temp.txt")
# Папка и её содержимое автоматически удаляются

Сложный пример с множественными контекстными менеджерами

with open(input.csv) as f_in, open(output.csv, w) as f_out:
    for line in f_in:
        f_out.write(process_line(line))
# Оба файла закроются при выходе

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

• Автоматическое управление ресурсами — даже при исключениях
• Чистый и читаемый код — явное начало и конец работы с ресурсом
• Предсказуемое поведение — гарантированная очистка
• Безопасность — защита от утечек ресурсов

Контекстные менеджеры — фундаментальный инструмент Python для надёжного управления ресурсами в Data Engineering и других областях разработки.