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

⏱ 35 мин · 🧩 заданий: 5

Проблема: ресурсы надо освобождать

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

f = open("data.txt")
try:
    data = f.read()
    process(data)          # если здесь исключение...
finally:
    f.close()              # ...файл всё равно закроется

with делает то же самое короче и надёжнее:

with open("data.txt") as f:
    data = f.read()
    process(data)
# здесь файл уже закрыт — при любом исходе

Протокол контекстного менеджера

with EXPR as VAR:
    BLOCK
  1. Вычисляется EXPR → менеджер mgr.
  2. Вызывается mgr.__enter__(); результат кладётся в VAR.
  3. Выполняется BLOCK.
  4. При любом выходе вызывается mgr.__exit__(exc_type, exc, traceback). Если исключения не было, все три аргумента — None.
  5. Если __exit__ вернул истину, исключение подавляется.
class Resource:
    def __init__(self, name):
        self.name = name

    def __enter__(self):
        print(f"  открываю {self.name}")
        return self                         # это попадёт в переменную после as

    def __exit__(self, exc_type, exc, tb):
        print(f"  закрываю {self.name}; исключение: {exc_type.__name__ if exc_type else None}")
        return False                        # не подавлять исключения

with Resource("БД") as r:
    print("  работаю с", r.name)

try:
    with Resource("файл"):
        raise ValueError("что-то сломалось")
except ValueError as e:
    print("поймали снаружи:", e)

Пройдите пример 👣 Пошагово: видно, что __exit__ вызывается и при исключении.

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

class IgnoreErrors:
    def __init__(self, *types):
        self.types = types

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        if exc_type is not None and issubclass(exc_type, self.types):
            print(f"  проглотил {exc_type.__name__}: {exc}")
            return True                     # подавить
        return False

with IgnoreErrors(KeyError, IndexError):
    [][5]
print("программа продолжается")

Не подавляйте всё подряд

__exit__, который всегда возвращает True, прячет ошибки — включая опечатки. Подавляйте только конкретные, ожидаемые исключения.

contextlib.contextmanager

Писать класс ради двух методов долго. Декоратор @contextmanager превращает генератор в контекстный менеджер: код до yield — это __enter__, значение yield — то, что попадёт в as, код после — __exit__:

from contextlib import contextmanager
import time

@contextmanager
def timer(label):
    start = time.perf_counter()
    try:
        yield                                  # здесь выполняется тело with
    finally:                                   # finally — чтобы сработало и при исключении
        print(f"{label}: {(time.perf_counter() - start) * 1000:.1f} мс")

with timer("сумма"):
    total = sum(range(200_000))
from contextlib import contextmanager

@contextmanager
def transaction(db):
    db.append("BEGIN")
    try:
        yield db
    except Exception:
        db.append("ROLLBACK")
        raise                                  # пробрасываем ошибку дальше
    else:
        db.append("COMMIT")

log = []
with transaction(log) as tx:
    tx.append("INSERT user")

try:
    with transaction(log) as tx:
        tx.append("INSERT order")
        raise ValueError("нет товара")
except ValueError:
    pass
print(log)

Забытый try/finally

Без try/finally вокруг yield код «выхода» не выполнится при исключении в блоке with — ресурс утечёт.

Готовые менеджеры из contextlib

from contextlib import suppress, redirect_stdout, nullcontext, closing
import io

with suppress(FileNotFoundError, KeyError):      # тихо игнорировать
    {}["нет"]

buf = io.StringIO()
with redirect_stdout(buf):                       # перехватить print
    print("это ушло в буфер")
print("перехвачено:", repr(buf.getvalue()))

def process(lock=None):
    with lock or nullcontext():                  # «пустой» менеджер вместо if
        return "готово"
print(process())

class Conn:
    def close(self):
        print("Conn.close()")
with closing(Conn()):                            # вызвать close() у объекта без with
    pass

ExitStack — переменное число менеджеров

from contextlib import ExitStack, contextmanager

@contextmanager
def opened(name):
    print(f"  открыт {name}")
    yield name
    print(f"  закрыт {name}")

files = ["a.txt", "b.txt", "c.txt"]
with ExitStack() as stack:
    handles = [stack.enter_context(opened(f)) for f in files]
    stack.callback(print, "  произвольная очистка")
    print("работаю с", handles)
# закрываются в обратном порядке

Несколько менеджеров

from contextlib import contextmanager

@contextmanager
def tag(name):
    print(f"<{name}>")
    yield
    print(f"</{name}>")

with tag("html"), tag("body"):        # эквивалент вложенных with
    print("  привет")

with (
    tag("ul"),                        # скобки для переноса строк — Python 3.10+
    tag("li"),
):
    print("  пункт")

Где встречается with

Код Что гарантирует
with open(path) as f: файл закроется
with lock: блокировка threading.Lock будет снята
with transaction.atomic(): (Django) коммит или откат транзакции
with Session() as s: (SQLAlchemy, requests, httpx) соединение вернётся в пул
with pytest.raises(ValueError): тест упадёт, если исключения не было
with mock.patch("mod.func"): подмена отменится после блока
with decimal.localcontext() as ctx: временная точность вычислений
with tempfile.TemporaryDirectory() as d: временная папка удалится
async with session.get(url) as resp: асинхронный вариант (__aenter__/__aexit__)

Вопрос на собеседовании

«Как написать свой контекстный менеджер?» — Двумя способами: класс с __enter__ и __exit__(exc_type, exc, tb) (вернуть True — подавить исключение) или генератор с одним yield, обёрнутый в @contextlib.contextmanager, где очистка стоит в finally.

Итог

  • with гарантирует вызов __exit__ при любом выходе из блока.
  • __enter__ возвращает значение для as; __exit__ получает информацию об исключении и может подавить его, вернув True.
  • @contextmanager — менеджер из генератора; очистку ставьте в finally.
  • suppress, redirect_stdout, nullcontext, closing, ExitStack закрывают типовые задачи.

Практика решено 0 из 5

Вы не вошли — ответы проверяются, но прогресс не сохранится. Войти или зарегистрироваться.
Задание 1 Программа на Python

Напишите класс-контекстный менеджер Timer, который при входе запоминает время (time.perf_counter()), а при выходе сохраняет длительность в атрибут elapsed. with Timer() as t: должен давать в t сам объект таймера. Исключения подавлять не нужно.

Какие тесты проверяют код (2)
import time
with Timer() as t:
    time.sleep(0.01)
assert isinstance(t, Timer) and 0.005 < t.elapsed < 1

t = Timer()
try:
    with t:
        raise KeyError("x")
    ok = False
except KeyError:
    ok = True
assert ok and hasattr(t, "elapsed"), "Исключение должно пролетать наружу, а elapsed — посчитаться"

Задание 2 Программа на Python

С помощью contextlib.contextmanager напишите temp_setting(config, key, value): внутри блока with в словаре config ключ key временно равен value, а после выхода (даже при исключении) восстанавливается старое значение. Если ключа раньше не было — после выхода его не должно быть.

Какие тесты проверяют код (3)
cfg = {"debug": False}
with temp_setting(cfg, "debug", True):
    assert cfg["debug"] is True
assert cfg == {"debug": False}

cfg = {}
with temp_setting(cfg, "mode", "test"):
    assert cfg["mode"] == "test"
assert cfg == {}, "Ключа не было — после выхода его быть не должно"

cfg = {"level": 1}
try:
    with temp_setting(cfg, "level", 99):
        raise RuntimeError
except RuntimeError:
    pass
assert cfg == {"level": 1}, "Значение должно восстановиться и при исключении"

Задание 3 Один вариант ответа

Что выведет код?

class Quiet:
    def __enter__(self):
        return self
    def __exit__(self, *args):
        return True

with Quiet():
    1 / 0
print("дальше")
Задание 4 Один вариант ответа

Что окажется в переменной f в with open("a.txt") as f:?

Задание 5 Один вариант ответа

Когда вызывается __exit__?

Войдите, чтобы отмечать пройденные уроки.