- Объяснять, как оператор
withвызывает__enter__и__exit__ - Писать класс контекстного менеджера, правильно обрабатывающий исключения
- Создавать менеджер на основе генератора с помощью
@contextmanager - Использовать готовые менеджеры:
suppress,redirect_stdout,ExitStack
Программа открывает файл, соединение с базой данных или сетевой сокет, а на полпути что-то идёт не так — возникает исключение или функция досрочно делает return. Если ресурс не освободить, файлы остаются заблокированными, соединения копятся, а данные могут так и не попасть на диск. С записью with open(...) ты познакомился в уроке о файлах; теперь посмотрим, что происходит у неё внутри, и напишем собственные контекстные менеджеры для любого ресурса, который нужно подготовить, а потом гарантированно освободить.
Что на самом деле делает with
Контекстный менеджер — объект с двумя методами. with manager as x: вызывает manager.__enter__() и связывает возвращённое значение с x. Затем выполняется блок. Что бы ни случилось дальше — блок завершился нормально, сработал return, break или возникло исключение, — Python вызывает manager.__exit__(exc_type, exc, tb). Файловый объект сам является своим контекстным менеджером, и его __exit__ закрывает файл:
with open('notes.txt', 'w', encoding='utf-8') as f:
f.write('first line\n')
print('inside:', f.closed)
print('after:', f.closed)
try:
with open('notes.txt', encoding='utf-8') as f:
text = f.read()
raise ValueError('something went wrong')
except ValueError as e:
print('error:', e)
print('closed even after the error:', f.closed)▸ Ожидаемый результат
inside: False after: True error: something went wrong closed even after the error: True
Python превращает оператор with примерно в такой код. Обрати внимание: __exit__ получает исключение и сам решает, распространяться ли ему дальше:
manager = open('notes.txt', encoding='utf-8')
f = manager.__enter__()
try:
text = f.read()
except BaseException as e:
if not manager.__exit__(type(e), e, e.__traceback__):
raise
else:
manager.__exit__(None, None, None)Это тот же шаблон try/finally, который иначе пришлось бы вручную писать в каждом месте. Контекстный менеджер упаковывает логику подготовки и очистки в один повторно используемый объект: код, работающий с ресурсом, остаётся коротким и просто не может «забыть» об очистке.
Собственный класс контекстного менеджера
Три аргумента __exit__ описывают исключение: его тип, сам объект исключения и трассировку (traceback). Если блок завершился нормально, все три равны None. Возвращаемое значение важно: истинное значение подавляет исключение, а False или None позволяют ему распространяться. Менеджер ниже восстанавливает словарь, если внутри блока что-то пошло не так, — как маленькая транзакция базы данных:
class Transaction:
def __init__(self, account):
self.account = account
def __enter__(self):
self.backup = dict(self.account)
return self.account
def __exit__(self, exc_type, exc, tb):
if exc_type is not None:
self.account.clear()
self.account.update(self.backup)
return False
wallet = {'Aysel': 100, 'Murad': 50}
for amount in (30, 500):
try:
with Transaction(wallet) as w:
w['Aysel'] -= amount
w['Murad'] += amount
if w['Aysel'] < 0:
raise ValueError('not enough money')
except ValueError as e:
print('cancelled:', e)
print(wallet)▸ Ожидаемый результат
{'Aysel': 70, 'Murad': 80}
cancelled: not enough money
{'Aysel': 70, 'Murad': 80}Первый перевод (30) прошёл. Второй сорвался на полпути: когда обнаружилась ошибка, 500 уже списали у Айсель и добавили Мураду, но __exit__ восстановил резервную копию. Вызывающий код всё равно видит исключение, потому что __exit__ вернул False.
contextlib.contextmanager: менеджер из генератора
Писать класс для каждого менеджера утомительно. Декоратор **@contextmanager** превращает функцию-генератор в контекстный менеджер: код до yield играет роль __enter__, значение после yield попадает в as, а код после yield играет роль __exit__. Если в блоке возникло исключение, оно повторно возбуждается внутри генератора прямо на строке yield — поэтому код очистки должен находиться в finally:
from contextlib import contextmanager
@contextmanager
def tag(name):
print(f'<{name}>')
try:
yield name
finally:
print(f'</{name}>')
with tag('ul'):
for item in ['tea', 'plov']:
with tag('li') as t:
print(f' {item} (inside {t})')▸ Ожидаемый результат
<ul> <li> tea (inside li) </li> <li> plov (inside li) </li> </ul>
Вывод показывает порядок: каждый li открывается и закрывается внутри ul, как вложенные скобки. Из менеджеров выходят в порядке, обратном входу, — последним вошёл, первым вышел.
Напиши контекстный менеджер timer(label), который печатает, сколько миллисекунд выполнялся код внутри блока with. Время должно печататься, даже если в блоке возникло исключение.
Показать решениеСкрыть решение
yield запомни время начала: start = time.perf_counter().perf_counter() — точные часы, предназначенные для измерения интервалов.Поставь
yield внутрь try, а разность time.perf_counter() - start вычисли в finally — тогда время напечатается и после ошибки.Блоку значение не нужно, поэтому достаточно просто
yield, а as не используется.import time
from contextlib import contextmanager
@contextmanager
def timer(label):
start = time.perf_counter()
try:
yield
finally:
elapsed = time.perf_counter() - start
print(f'{label}: {elapsed * 1000:.1f} ms')
with timer('sum of squares'):
total = sum(n * n for n in range(1_000_000))
print(total)Полезные менеджеры из стандартной библиотеки
В модуле contextlib есть несколько готовых менеджеров. suppress(Error) игнорирует перечисленные исключения, а redirect_stdout(buffer) временно перенаправляет вывод print в другой поток:
from contextlib import suppress, redirect_stdout
import io
import os
with suppress(FileNotFoundError):
os.remove('no-such-file.txt')
print('no crash')
buffer = io.StringIO()
with redirect_stdout(buffer):
print('this goes into the buffer')
print('captured:', buffer.getvalue().strip())▸ Ожидаемый результат
no crash captured: this goes into the buffer
Если число ресурсов известно только во время работы программы, используй **ExitStack**: он собирает любое количество менеджеров и в конце блока выходит из всех в обратном порядке — даже если открыть третий файл не удалось:
from contextlib import ExitStack
names = ['a.txt', 'b.txt', 'c.txt']
for i, name in enumerate(names):
with open(name, 'w', encoding='utf-8') as f:
f.write(f'file {i}\n')
with ExitStack() as stack:
files = [stack.enter_context(open(n, encoding='utf-8')) for n in names]
print([f.readline().strip() for f in files])
print(all(f.closed for f in files))▸ Ожидаемый результат
['file 0', 'file 1', 'file 2'] True
С with работают и многие другие объекты. Когда что-то настраивается временно, ищи контекстный менеджер:
| Объект | Что происходит при выходе |
|---|---|
open(...) | файл закрывается |
threading.Lock() | блокировка снимается |
decimal.localcontext() | восстанавливается прежняя точность |
tempfile.TemporaryDirectory() | папка удаляется вместе с содержимым |
unittest.mock.patch(...) | возвращается исходный объект |
sqlite3.connect(...) | транзакция фиксируется или откатывается (соединение не закрывается!) |
from decimal import Decimal, localcontext
print(Decimal(1) / Decimal(7))
with localcontext() as ctx:
ctx.prec = 5
print(Decimal(1) / Decimal(7))
print(Decimal(1) / Decimal(7))▸ Ожидаемый результат
0.1428571428571428571428571429 0.14286 0.1428571428571428571428571429
Допиши контекстный менеджер section(title): при входе он печатает == title ==, а при выходе — -- end of title --. Строка выхода должна печататься, даже если в блоке возникло исключение, а само исключение должно дойти до вызывающего кода.
from contextlib import contextmanager
@contextmanager
def section(title):
# print '== title ==' on entry and '-- end of title --' on exit, even after an error
yield
with section('Report'):
print('all good')
try:
with section('Import'):
raise ValueError('bad file')
except ValueError as e:
print('error:', e)▸ Ожидаемый результат
== Report == all good -- end of Report -- == Import == -- end of Import -- error: bad file
Исправь метод __exit__ класса Ignore(*exceptions): если исключение в блоке относится к одному из заданных типов, подави его; иначе (и когда исключения нет) верни False.
class Ignore:
def __init__(self, *exceptions):
self.exceptions = exceptions
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
# return True only if the exception is one of self.exceptions
return False
with Ignore(ZeroDivisionError):
print(1 / 0)
print('still running')
with Ignore(KeyError, IndexError):
[][5]
print('done')▸ Ожидаемый результат
still running done
Главное
with m as xвызываетm.__enter__()(результат попадает вx), а в конце обязательно вызываетсяm.__exit__().__exit__(exc_type, exc, tb)получает сведения об исключении; истинное возвращаемое значение подавляет его.- В
@contextmanagerкод доyield— вход, после — выход; очистка должна быть вfinally. - Из нескольких менеджеров выходят в обратном порядке; для их динамического числа есть
ExitStack. suppress,redirect_stdout,localcontextиTemporaryDirectory— готовые контекстные менеджеры.
Проверь себя
Вопросов: 10. Каждый правильный ответ приносит XP.
__exit__?