Каждый разработчик знает with open("file.txt") as f. Файл открывается, читается, закрывается автоматически, даже если внутри блока произошла ошибка. Удобно, понятно, да и безопасно. Но почему-то with почти всегда встречается только рядом с файлами. А ведь контекстные менеджеры решают гораздо более широкий класс задач: управление соединениями с базой, транзакции, замер времени, временное изменение конфигурации, подавление ошибок, захват и освобождение ресурсов.

Написать свой контекстный менеджер — дело на 5-10 строк, и код после этого становится заметно чище.

Что на самом деле делает with

with вызывает два метода: enter при входе в блок и exit при выходе (в том числе при исключении). Всё, что нужно от контекстного менеджера, — реализовать эти два метода:

class Timer:
    def __enter__(self):
        self.start = time.perf_counter()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.perf_counter() - self.start
        print(f"Elapsed: {self.elapsed:.3f}s")
        return False  # не подавляем исключения

with Timer() as t:
    result = expensive_computation()

print(f"Computation took {t.elapsed:.3f}s")

enter возвращает объект, который попадает в переменную после as. exit получает информацию об исключении (если оно было) и решает, подавлять его или нет (return True подавляет, return False пробрасывает дальше).

Но писать класс ради двух методов — многословно. В стандартной библиотеке есть contextlib.contextmanager, который позволяет написать то же самое как генератор:

from contextlib import contextmanager
import time

@contextmanager
def timer(label: str = "Block"):
    start = time.perf_counter()
    try:
        yield  # здесь выполняется тело with-блока
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.3f}s")

with timer("DB query"):
    result = db.execute("SELECT * FROM orders")

Код до yield — это enter. Код после yield (в finally) — это exit. Если нужно вернуть значение через as, делаете yield value. Проще, короче, и для большинства случаев достаточно.

Управление соединениями с базой данных

Типичный паттерн: получить соединение из пула, выполнить работу, вернуть соединение в пул.

Если забыть вернуть — соединение утекает, пул исчерпается, сервис встанет.

@contextmanager
def get_connection(pool):
    conn = pool.getconn()
    try:
        yield conn
    finally:
        pool.putconn(conn)

# Использование
with get_connection(db_pool) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE active = true")
    users = cursor.fetchall()
# Соединение вернулось в пул, даже если был exception

Без контекстного менеджера код выглядит так:

conn = pool.getconn()
try:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE active = true")
    users = cursor.fetchall()
finally:
    pool.putconn(conn)

Формально разница невелика. Но когда у вас 50 мест в коде, где нужно соединение, try/finally на каждом — это шум, который мешает читать бизнес-логику. С контекстным менеджером управление ресурсом спрятано в одном месте, и в каждом месте использования — только with get_connection(pool) as conn.

Транзакции

Соединение + транзакция — частая комбинация. Начинаем транзакцию, если всё хорошо — commit, если exception — rollback:

@contextmanager
def transaction(pool):
    conn = pool.getconn()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise  # пробрасываем exception дальше
    finally:
        pool.putconn(conn)

with transaction(db_pool) as conn:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
# Если второй UPDATE упал — rollback обоих

Без контекстного менеджера это try/except/finally на 10 строк в каждом месте, где нужна транзакция. С ним — одна строка with transaction(pool) as conn, и дальше только бизнес-логика.

Обратите внимание на raise в блоке except. Без него исключение будет подавлено, и вызывающий код не узнает, что транзакция откатилась. Контекстный менеджер делает cleanup (rollback), но решение о том, что делать с ошибкой, оставляет вызывающему коду.

Замер времени с контекстом

Timer из начала статьи полезен, но базовый. Версия, которая пригодится:

@contextmanager
def measure(operation: str, logger=None, warn_threshold: float = None):
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        msg = f"{operation}: {elapsed:.3f}s"
        
        if logger:
            if warn_threshold and elapsed > warn_threshold:
                logger.warning(f"SLOW {msg}")
            else:
                logger.info(msg)
        else:
            print(msg)

with measure("fetch_users", logger=log, warn_threshold=1.0):
    users = db.fetch_all_users()

with measure("process_payment", logger=log, warn_threshold=0.5):
    process_payment(order)

Если операция заняла дольше warn_threshold, в лог пойдёт WARNING вместо INFO. В продакшене можно настроить алерт на WARNING-логи и узнавать о деградации производительности до того, как пользователи начнут жаловаться.

Этот менеджер можно комбинировать с structured logging (structlog / slog): вместо logger.info(msg) передавать поля operation и duration_ms отдельно, и фильтровать в Kibana.

Временное изменение состояния

Иногда нужно временно изменить что-то (рабочую директорию, переменные окружения, уровень логирования, настройки) и гарантированно вернуть как было:

import os

@contextmanager
def working_directory(path: str):
    original = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(original)

with working_directory("/tmp/build"):
    subprocess.run(["make", "all"])
# Вернулись в оригинальную директорию

@contextmanager
def env_var(key: str, value: str):
    original = os.environ.get(key)
    os.environ[key] = value
    try:
        yield
    finally:
        if original is None:
            del os.environ[key]
        else:
            os.environ[key] = original

with env_var("DATABASE_URL", "postgres://test:test@localhost/testdb"):
    run_tests()
# DATABASE_URL вернулся к прежнему значению (или удалён, если его не было)

Без контекстного менеджера нужно руками сохранять старое значение, менять, потом восстанавливать в finally. С ним — одна строка, и невозможно забыть восстановить.

Для тестов это особенно ценно. Тест меняет переменную окружения, запускает код, переменная восстанавливается. Следующий тест не зависит от предыдущего.

Подавление исключений

В стандартной библиотеке есть contextlib.suppress:

from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove("temp_file.txt")
# Если файла нет — ничего не произойдёт, исключение подавлено

Но можно написать свой менеджер, который не просто подавляет, а логирует:

@contextmanager
def ignore_errors(*exceptions, logger=None):
    try:
        yield
    except exceptions as e:
        if logger:
            logger.warning(f"Suppressed {type(e).__name__}: {e}")

with ignore_errors(ConnectionError, TimeoutError, logger=log):
    send_analytics_event(data)
# Если аналитика недоступна — ничего страшного, основная логика продолжит работу

Полезно для некритичных операций: отправка аналитики, обновление кеша, логирование во внешний сервис. Если они упали, основная логика не должна страдать, но знать об ошибке нужно.

Async-контекстные менеджеры

Для async-кода всё то же самое, но с async with, aenter/__aexit__ и asynccontextmanager:

from contextlib import asynccontextmanager

@asynccontextmanager
async def async_transaction(pool):
    conn = await pool.acquire()
    tr = conn.transaction()
    await tr.start()
    try:
        yield conn
        await tr.commit()
    except Exception:
        await tr.rollback()
        raise
    finally:
        await pool.release(conn)

async with async_transaction(db_pool) as conn:
    await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")

Принцип идентичен синхронной версии. asynccontextmanager вместо contextmanager, async def вместо def, await где нужно.

Когда не стоит писать контекстный менеджер

Если cleanup-логика нужна в одном месте, try/finally проще и читаемее: не нужно уходить в отдельную функцию, чтобы понять, что происходит. Контекстный менеджер окупается, когда один и тот же паттерн (получить ресурс, использовать, вернуть) повторяется в трёх и более местах.

Если cleanup зависит от результата работы внутри блока (не просто «вернуть соединение», а «если результат X — сделать Y, если Z — сделать W»), контекстный менеджер становится неудобным. exit получает информацию об исключении, но не о результате нормальной работы. В таких случаях проще обычный try/except/finally с явной логикой.

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

Кстати, если хотите глубже разобраться в теме backend-инфраструктуры, async-кода и работы с данными, обратите внимание на бесплатные открытые уроки OTUS:

• 2 июня, 20:00 — «Polyglot Persistence: как современные системы живут с десятками баз данных».
  _{Как современные backend-системы работают сразу с несколькими типами БД и зачем это нужно в реальных проектах.}

• 16 июня, 20:00 — «Асинхронная обработка данных в высоконагруженных системах».
  _{Про async-подходы, очереди, параллельную обработку и архитектурные решения под высокую нагрузку.}

Занятия проходят в рамках онлайн-курсов OTUS и позволяют познакомиться с преподавателями-практиками и форматом обучения.

Больше анонсов открытых уроков, подборок по IT-направлениям и полезных материалов публикуем на канале OTUS в MAX — подписывайтесь.