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

Материал из llmdocs — справочник разработчика
Портал: Python
Контекстные менеджеры
Синтаксис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, где освобождение привязано к владению.

См. также

Источники

  1. PEP 343 — The "with" Statement
  2. Python Language Reference — The with statement
  3. Python — contextlib: Utilities for with-statement contexts
  4. Python Data Model — With Statement Context Managers
  5. PEP 492 — Coroutines with async and await syntax (async with)
  6. Python — unittest.mock.patch