Контекстные менеджеры и with
Проблема: ресурсы надо освобождать
Файлы, сетевые соединения, блокировки, транзакции нужно закрывать, даже если по пути случилась ошибка. Без 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
- Вычисляется
EXPR→ менеджерmgr. - Вызывается
mgr.__enter__(); результат кладётся вVAR. - Выполняется
BLOCK. - При любом выходе вызывается
mgr.__exit__(exc_type, exc, traceback). Если исключения не было, все три аргумента —None. - Если
__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
Напишите класс-контекстный менеджер 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 — посчитаться"
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
return False
С помощью 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}, "Значение должно восстановиться и при исключении"
from contextlib import contextmanager
_missing = object()
@contextmanager
def temp_setting(config, key, value):
old = config.get(key, _missing)
config[key] = value
try:
yield config
finally:
if old is _missing:
del config[key]
else:
config[key] = old
Что выведет код?
class Quiet:
def __enter__(self):
return self
def __exit__(self, *args):
return True
with Quiet():
1 / 0
print("дальше")
Что окажется в переменной f в with open("a.txt") as f:?
Когда вызывается __exit__?
Войдите, чтобы отмечать пройденные уроки.