Контекстные менеджеры
| Синтаксис | with expr as name — блок |
|---|---|
| Протокол | enter и exit (async — aenter и aexit) |
| Введены | Python 2.5, PEP 343 (2006) |
| Модуль-помощник | contextlib |
| Типичные применения | файлы, блокировки, транзакции, временные изменения, тайм-ауты |
| Аналоги в других языках | try-with-resources (Java), using (C#), defer (Go), RAII и Drop (C++, Rust) |
Контекстный менеджер — объект Python, который определяет, что происходит при входе в блок with и при выходе из него. Главная задача — гарантированно освободить ресурс (закрыть файл, вернуть соединение в пул, отпустить блокировку, откатить транзакцию), как бы ни завершился блок: нормально, через return или с исключением. Оператор with появился в Python 2.5 по PEP 343.[1]
Зачем нужен
Без контекстного менеджера освобождение ресурса приходится писать в try/finally:
f = open("data.txt", encoding="utf-8")
try:
text = f.read()
finally:
f.close()
С ним то же самое короче и надёжнее — забыть закрытие невозможно:
with open("data.txt", encoding="utf-8") as f:
text = f.read()
# здесь файл уже закрыт
Несколько менеджеров можно объединить в одном with; с Python 3.10 официально поддерживаются скобки для переноса строк:[2]
with (
open("in.csv") as src,
open("out.csv", "w") as dst,
):
dst.write(src.read())
Протокол
Контекстный менеджер — любой объект с двумя методами:[4]
__enter__(self)— вызывается при входе; его результат присваивается переменной послеas;__exit__(self, exc_type, exc, tb)— вызывается при выходе; если блок завершился исключением, аргументы описывают его, иначе все три равныNone. Если метод вернёт истинное значение, исключение будет подавлено.
import time
class Timer:
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, exc_type, exc, tb):
self.elapsed = time.perf_counter() - self.start
print(f"{self.elapsed:.3f} с")
return False # исключения не глотаем
with Timer() as t:
sum(range(10_000_000))
contextlib
Модуль contextlib избавляет от написания классов.[3] Декоратор @contextmanager превращает генератор в контекстный менеджер: код до yield — вход, после — выход.
from contextlib import contextmanager
import os
@contextmanager
def chdir_tmp(path):
old = os.getcwd()
os.chdir(path)
try:
yield path
finally:
os.chdir(old) # выполнится и при исключении
try/finally вокруг yield обязателен — иначе при исключении в блоке восстановление не выполнится. Сам @contextmanager — пример декоратора.
Другие полезные инструменты:
| Инструмент | Что делает |
|---|---|
suppress(FileNotFoundError) |
подавляет указанные исключения |
closing(obj) |
вызывает obj.close() у объектов без протокола |
redirect_stdout(buf) |
временно перенаправляет вывод |
chdir(path) (3.11+) |
временно меняет рабочий каталог |
nullcontext() |
«пустой» менеджер для условного кода |
ExitStack |
динамический набор менеджеров, например список файлов неизвестной длины |
asynccontextmanager |
то же, что @contextmanager, для асинхронного кода |
from contextlib import ExitStack
with ExitStack() as stack:
files = [stack.enter_context(open(p)) for p in paths]
... # все файлы закроются в обратном порядке
Асинхронные контекстные менеджеры
В asyncio вход и выход могут требовать ожидания — например, открыть сетевое соединение. Для этого есть async with и методы __aenter__/__aexit__ (PEP 492):[5]
import asyncio, httpx
async def main():
async with httpx.AsyncClient() as client:
r = await client.get("https://example.com")
async with asyncio.timeout(5): # Python 3.11+
await slow_operation()
async with asyncio.TaskGroup() as tg: # структурная конкурентность
tg.create_task(job(1))
tg.create_task(job(2))
Где встречаются
- Файлы и сокеты —
open(),socket.socket,zipfile.ZipFile,tempfile.TemporaryDirectory(). - Блокировки —
with threading.Lock():захватывает и гарантированно отпускает мьютекс (см. Состояние гонки и блокировки). - Транзакции БД — в
sqlite3соединение как менеджер фиксирует транзакцию при успехе и откатывает при исключении; аналогичноwith session.begin():в SQLAlchemy (см. Транзакция и ACID). - Тесты —
pytest.raises(ValueError)проверяет, что блок бросил исключение,unittest.mock.patchвременно подменяет объект (см. pytest).[6] - Временные настройки —
decimal.localcontext(),warnings.catch_warnings(),np.errstate()в NumPy,torch.no_grad()в PyTorch.
Подводные камни
- Объект, созданный в
with, после блока остаётся доступен по имени, но ресурс уже закрыт — повторное чтение закрытого файла дастValueError. - Возвращать
Trueиз__exit__нужно осознанно: случайное подавление исключений прячет ошибки. - Генератор в
@contextmanagerдолжен выдаватьyieldровно один раз. - Менеджер обычно одноразовый; переиспользуемые (
Lock) и повторно входимые (RLock,ExitStack) — исключения.
Аналоги в других языках
Идея «ресурс освобождается при выходе из области видимости» реализована по-разному: try-with-resources в Java, using в C#, defer в Go, RAII в C++ и трейт Drop в Rust, где освобождение привязано к владению.