Каждый разработчик знает 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 — подписывайтесь.