Какова основная задача контекстного менеджера

Контекстные менеджеры в Python: основная задача и применение

Контекстные менеджеры - один из самых мощных и недооцененных инструментов Python. Это механизм для гарантированного управления ресурсами.

Основная задача контекстного менеджера

Гарантированное выполнение cleanup кода, даже при ошибках.

# Проблема БЕЗ контекстного менеджера
file = open('data.txt', 'r')
try:
    data = file.read()
    # Если здесь возникнет исключение...
    process_data(data)  # ValueError!
finally:
    file.close()  # Это выполнится ДА

# vs

# С контекстным менеджером
with open('data.txt', 'r') as file:
    data = file.read()
    process_data(data)  # ValueError!
# file.close() выполнится автоматически, даже при ошибке

Встроенные контекстные менеджеры

1. open() - Управление файлами

# ✓ Гарантированно закроет файл
with open('users.txt', 'r') as f:
    for line in f:
        print(line.strip())
# Даже если будет ошибка во время чтения - файл закроется

# Эквивалент без with (плохо):
f = open('users.txt', 'r')
try:
    for line in f:
        print(line.strip())
finally:
    f.close()  # Можно забыть!)

2. threading.Lock() - Синхронизация потоков

import threading

shared_resource = 0
lock = threading.Lock()

# ✓ Гарантированное освобождение локи
with lock:
    shared_resource += 1
    # Даже при ошибке - лок освободится и другой поток может выполниться

# Эквивалент:
lock.acquire()
try:
    shared_resource += 1
finally:
    lock.release()

3. contextlib.contextmanager() - Создание своих менеджеров

from contextlib import contextmanager
import time

@contextmanager
def timer(name: str):
    """Контекстный менеджер для измерения времени"""
    print(f"[{name}] Starting...")
    start = time.time()
    try:
        yield  # Уступляем управление блоку with
    finally:
        elapsed = time.time() - start
        print(f"[{name}] Completed in {elapsed:.2f}s")

# Использование
with timer("Database query"):
    # Это будет измерено
    time.sleep(2)
    # Вывод:
    # [Database query] Starting...
    # [Database query] Completed in 2.00s

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

Способ 1: Класс с enter и exit

class DatabaseConnection:
    """Контекстный менеджер для БД подключения"""
    
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        # Код, выполняемый при входе в блок with
        print(f"Connecting to {self.connection_string}...")
        self.connection = self._establish_connection()
        return self.connection  # Это будет доступно как переменная after as
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Код, выполняемый при выходе из блока with
        # exc_type, exc_val, exc_tb содержат информацию об ошибке, если она была
        
        if exc_type is not None:
            print(f"Error occurred: {exc_type.__name__}: {exc_val}")
            self.connection.rollback()  # Откатываем транзакцию
        else:
            self.connection.commit()  # Коммитим если успешно
        
        print("Closing connection...")
        self.connection.close()
        
        # Возвращаем False если хотим распространить ошибку
        # Возвращаем True если хотим подавить ошибку
        return False
    
    def _establish_connection(self):
        # Имитация подключения
        return MockConnection()

# Использование
with DatabaseConnection("postgres://localhost/mydb") as db:
    # db - это объект, возвращенный __enter__
    result = db.execute("SELECT * FROM users")
    print(result)

# Вывод:
# Connecting to postgres://localhost/mydb...
# Closing connection...

Способ 2: Использование @contextmanager декоратора

from contextlib import contextmanager
import logging

@contextmanager
def logger_context(name: str, level=logging.INFO):
    """Контекстный менеджер для логирования"""
    logger = logging.getLogger(name)
    
    # __enter__
    logger.log(level, f"Entering {name}")
    start_time = time.time()
    
    try:
        yield logger  # Передаем logger в блок with
    except Exception as e:
        # __exit__ - обработка ошибки
        logger.error(f"Error in {name}: {e}", exc_info=True)
        raise  # Пробросить ошибку дальше
    finally:
        # __exit__ - гарантированная очистка
        elapsed = time.time() - start_time
        logger.log(level, f"Exiting {name} (took {elapsed:.2f}s)")

# Использование
with logger_context("DataProcessing") as log:
    log.info("Processing started")
    # Обработка...
    log.info("Processing completed")

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

1. Управление базой данных (async version)

from contextlib import asynccontextmanager
import asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

@asynccontextmanager
async def get_db_session():
    """Контекстный менеджер для async БД сессии"""
    engine = create_async_engine(
        'postgresql+asyncpg://user:password@localhost/mydb'
    )
    async with AsyncSession(engine) as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await engine.dispose()

# Использование в FastAPI
@app.get('/users/{user_id}')
async def get_user(user_id: int):
    async with get_db_session() as db:
        user = await db.query(User).filter(User.id == user_id).first()
        return user

2. Управление временем выполнения и логирование

from contextlib import contextmanager
import time
import logging
from functools import wraps

@contextmanager
def measure_time(operation_name: str):
    """Измеряет время выполнения блока кода"""
    logger = logging.getLogger(__name__)
    start = time.time()
    
    try:
        yield
    finally:
        elapsed = time.time() - start
        logger.info(f"{operation_name} took {elapsed:.3f}s")
        
        # Можно добавить метрики
        if elapsed > 1.0:
            logger.warning(f"{operation_name} is slow! ({elapsed:.3f}s)")

# Использование
with measure_time("Database query"):
    result = db.execute("SELECT * FROM huge_table")

3. Управление ресурсами в многопроцессности

from contextlib import contextmanager
from multiprocessing import Pool

@contextmanager
def get_process_pool(num_workers: int = 4):
    """Контекстный менеджер для пула процессов"""
    pool = Pool(num_workers)
    try:
        yield pool
    finally:
        pool.close()  # Остановить принятие новых заданий
        pool.join()   # Дождаться завершения всех

# Использование
with get_process_pool(8) as pool:
    results = pool.map(expensive_function, data)
    # Pool автоматически очистится

4. Управление состоянием приложения

from contextlib import contextmanager
from enum import Enum

class AppState(Enum):
    RUNNING = "running"
    PAUSED = "paused"
    STOPPED = "stopped"

class Application:
    def __init__(self):
        self.state = AppState.STOPPED
    
    @contextmanager
    def pause_application(self):
        """Временно остановить приложение"""
        old_state = self.state
        self.state = AppState.PAUSED
        print(f"Paused: {old_state} -> {AppState.PAUSED}")
        
        try:
            yield
        finally:
            self.state = old_state
            print(f"Resumed: {AppState.PAUSED} -> {old_state}")

# Использование
app = Application()
app.state = AppState.RUNNING

with app.pause_application():
    # В этом блоке app паузирован
    print(f"Current state: {app.state}")  # PAUSED

print(f"Current state: {app.state}")  # RUNNING

Встроенные полезные контекстные менеджеры из contextlib

from contextlib import (
    contextmanager,      # Декоратор для создания менеджера
    asynccontextmanager, # Для async функций
    closing,             # Гарантирует вызов close()
    suppress,            # Подавляет исключения
    redirect_stdout,     # Перенаправляет stdout
    redirect_stderr,     # Перенаправляет stderr
    ExitStack            # Управление множественными контекстами
)

# Пример 1: closing() - для объектов с close() методом
with closing(open('file.txt')) as f:
    data = f.read()

# Пример 2: suppress() - подавляет исключения
with suppress(FileNotFoundError):
    os.remove('nonexistent.txt')  # Не выбросит ошибку

# Пример 3: redirect_stdout() - перенаправляет вывод
from io import StringIO

buffer = StringIO()
with redirect_stdout(buffer):
    print("This goes to buffer")
    
output = buffer.getvalue()  # "This goes to buffer\n"

# Пример 4: ExitStack() - управление множественными контекстами
with ExitStack() as stack:
    file1 = stack.enter_context(open('file1.txt'))
    file2 = stack.enter_context(open('file2.txt'))
    file3 = stack.enter_context(open('file3.txt'))
    
    # Все три файла откроются, и при выходе все закроются
    # В обратном порядке: file3, file2, file1

Правила и best practices

class ContextManagerBestPractices:
    
    # 1. Всегда используй контекстные менеджеры для ресурсов
    @staticmethod
    def rule_1():
        # ✓ Хорошо
        with open('file.txt') as f:
            data = f.read()
        
        # ❌ Плохо
        f = open('file.txt')
        data = f.read()
        f.close()  # Могу забыть
    
    # 2. Всегда обрабатывай исключения в __exit__
    @staticmethod
    def rule_2():
        class GoodContextManager:
            def __enter__(self):
                return self
            
            def __exit__(self, exc_type, exc_val, exc_tb):
                # exc_type is None если нет ошибки
                if exc_type is not None:
                    # Обработать ошибку
                    pass
                return False  # Пробросить ошибку дальше
    
    # 3. Вложенные контекстные менеджеры
    @staticmethod
    def rule_3():
        # ✓ Правильно
        with open('in.txt') as fin, open('out.txt', 'w') as fout:
            for line in fin:
                fout.write(line.upper())
        
        # Оба файла закроются в правильном порядке
    
    # 4. Используй contextlib для простых случаев
    @staticmethod
    def rule_4():
        from contextlib import contextmanager
        
        @contextmanager
        def simple_manager():
            # Setup
            print("Enter")
            try:
                yield
            finally:
                # Cleanup
                print("Exit")
        
        # Проще чем создавать класс

Сравнение подходов

Вывод

Основная задача контекстного менеджера: гарантировать выполнение cleanup кода независимо от того, что происходит внутри блока with - успех, ошибка, return, break, continue. Это делает код безопаснее и читабельнее.

# Remember:
# with statement = гарантия cleanup
# Используй везде где управляются ресурсы:
# - Файлы
# - БД соединения
# - Блокировки
# - Временные переменные состояния
# - Профилирование/логирование