Контекстные менеджеры в Python
Контекстные менеджеры – одна из основных функций языка, которая делает Python уникальным. Оператор with позволяет разработчикам писать свой код в сжатом и понятном виде. Новый вложенный блок даёт визуальный сигнал и облегчает понимание кода. Понимание контекстных менеджеров является ключом к пониманию идеи питонического кода.
Контекстные менеджеры обычно используются для захвата и освобождения ресурсов, но это не единственный случай их использования. Они полезны для унификации общего кода настройки и разрыва или любой пары операций, которые необходимо выполнить до или после действия.
В этой статье будет рассказано о некоторых интересных примерах их использования в реальном коде и автор надеется, что статья станет стимулом для продолжения исследования этой темы.
Коротко о контекстных менеджерах
Говоря простыми словами, контекстные менеджеры упрощают запись блоков try-finally.
# Python код с общей настройкой и завершением:
setup()
try:
do_work()
finally:
teardown()
# может быть переписано на:
cm = ContextManager()
obj = cm.__enter__()
try:
do_work(obj) # используйте obj по мере необходимости (см. примеры ниже)
finally:
cm.__exit__()
# это может быть реализовано более лаконично с помощью оператора `with`:
with ContextManager() as obj:
do_work()
Выражение, следующее за ключевым словом with, должно возвращать объект, соответствующий протоколу Context Manager. Это может быть экземпляр класса или вызов функции, который возвращает объект Context Manager, а также реализовывать два специальных метода: enter и exit.
Некоторые важные моменты, которые следует помнить.
- enter должен возвращать объект, который присваивается переменной после as. По умолчанию это None и необязателен. Обычно возвращается self и сохранение функциональности, требуемую в рамках одного класса.
- exit вызывается в исходном объекте Context Manager, а не в объекте, возвращаемом enter.
- Если ошибка возникает в init или enter, тогда блок кода никогда не выполняется и exit не вызывается.
- После ввода в блок кода, exit вызывается всегда, даже если в нём возникает исключение.
- Если exit возвращает True, исключение будет подавлено.
class ContManager(object): def __init__(self): print('__init__') def __enter__(self): print('__enter__') return self def __exit__(self, type, value, traceback): print('__exit__:', type, value) return True # Подавить исключение def __del__(self): print('__del__', self) with ContManager() as c: print('Делаем что-нибудь с c:', c) raise RuntimeError() print('завершаем действия') print('Выполнено')
Вывод:
__init__
__enter__
Делаем что-нибудь с: <__main__.ContManager object at 0x116fd8af1>
__exit__: <type 'exceptions.RuntimeError'>
Выполнено
Простые менеджеры контекста также могут быть написаны с использованием генераторов и декоратора contextmanager:
from contextlib import contextmanager
@contextmanager
def context_manager_func():
setup()
# yield должен быть завернут в try / finally
# перехватывая любые исключения, возникающие в вызывающем коде
try:
yield
finally:
teardown()
Надежные деструкторы
Контекстные менеджеры дают нам надежный метод очистки ресурсов. В отличие от других языков OO, таких как C ++ и Java, вызов метода деструктора Python del не всегда гарантируется. Он вызывается только тогда, когда счётчик ссылок на объект достигает нуля. Это может произойти в конце текущей функции или в конце программы или никогда в случае циклических ссылок.
До Python 3.4, если все объекты в эталонном цикле имеют метод del, Python не будет вызывать его при их удалении сборщиком мусора. Это связано с тем, что у Python нет безопасного способа узнать, какой порядок удаления этих объекты.
Начиная с Python 3.4, объекты с del теперь могут быть удалены сборщиком мусора. Однако порядок их вызова не определён.
Давайте рассмотрим несколько реальных примеров с контекстными менеджерами.
Убедитесь, что открытый поток закрывается
Функция open() - это канонический пример менеджеров контекста. Он предоставляет дескриптор открытого файла, который будет закрыт в конце блока.
with open(path, mode) as f:
f.read()
Объекты StringIO ведут себя одинаково:
with io.StringIO() as b:
b.write('foo’)
Менеджер контекста closing вызывает метод close() для любого объекта, если такой метод в нём существует.
from contextlib import closing
with closing(urllib.urlopen('http://www.severcart.org')) as page:
page.readlines()
Проверка на возникновение исключения при тестировании
В unittest (2.7+):
def test_split(self):
with self.assertRaises(TypeError):
'hello world'.split(2)
with self.assertRaisesRegexp(TypeError, 'expected a string.*object$'):
'hello world'.split(2)
В Pytest
with pytest.raises(ValueError):
int('hello')
Настройка mocks перед тестированием
with mock.patch('a.b'):
call()
mock.patch - пример менеджера контекста, который также может использоваться в качестве декоратора. Python 3 поставляется с утилитой ContextDecorator, которая упрощает запись менеджеров контекста.
При использовании в качестве декоратора mock.patch передает вновь созданный mock объект (возвращаемое значение enter) в декорированную функцию. ContextDecorator в Python 3, с другой стороны, не предоставляет доступа к возвращаемому значению метода enter.
Синхронизация доступа к общим ресурсам
with threading.RLock():
access_resource()
Оператор with в этом случае вызывает lock.acquire() при входе и lock.release() при выходе.
В потоковом модуле в качестве контекстных менеджеров могут использоваться Lock, RLock, Condition, Semaphore, BoundedSemaphore.
Аналогичный подход можно использовать для блокировки файлов при доступе к ним. Например, диспетчер контекста pidfile использует fcntl.flock() для получения блокировки файла в python-демонах.
import fcntl, csv, sys
class open_locked:
def __init__(self, *args, **kwargs):
self.fd = open(*args, **kwargs)
def __enter__(self):
fcntl.flock(self.fd, fcntl.LOCK_EX)
return self.fd.__enter__()
def __exit__(self, type, value, traceback):
fcntl.flock(self.fd, fcntl.LOCK_UN)
return self.fd.__exit__()
with open_locked('data.csv', 'w') as outf:
writer = csv.writer(outf)
writer.writerows(someiterable)
Настройка среды выполнения Python
with decimal.localcontext() as ctx:
# настройка точности операции внутри этого контекста
ctx.prec = 42
math_operations()
Управление подключениями к базе данных и транзакциями
conn = sqlite3.connect(':memory:')
with conn:
# начинаем новую db
conn.execute('create table mytable (id int primary key, name char(50))')
conn.execute('insert into mytable(id) values (?)', (1, 'avni'))
# если что-то пошло не так, то таблица в БД создана не будет
# транзакция вернёт БД к исходному состоянию
...
# conn.commit() вызов.
Обёртка соединений по протоколу
class SomeProtocol:
def __init__(self, host, port):
self.host, self.port = host, port
def __enter__(self):
self._client = socket()
self._client.connect((self.host, self.port))
return self
def __exit__(self, exception, value, traceback):
self._client.close()
def send(self, payload): ...
def receive(self): ...
with SomeProtocol(host, port) as protocol:
protocol.send(['get', signal])
result = protocol.receive()
Тайминги выполнения кода
import time
class Timer:
def __init__(self, name):
self.name = name
def __enter__(self):
self.start = time.time()
def __exit__(self, *args):
self.end = time.time()
self.interval = self.end - self.start
print("%s took: %0.3f seconds" % (self.name, self.interval))
return False
>>> with Timer('Запрос домашней страницы'):
... conn = httplib.HTTPConnection('www.severcart.org')
... conn.request('GET', '/')
...
fetching severcart homepage took: 0.047 seconds
Автоматизация задач администрирования с использованием Fabric
Fabric предоставляет множество интересных менеджеров контекстов для автоматизации развертывания, выполнения локальных и удаленных задач.
# В Python 2.7+ несколько контекстных менеджеров можно комбинировать с помощью запятых
with cd('/path/to/app'), prefix('workon myvenv'):
run('./manage.py syncdb')
run('./manage.py loaddata myfixture')
Работа с временными файлами
import tempfile
with tempfile.NamedTemporaryFile() as tf:
print('Writing to tempfile:', tf.name)
tf.write('Some data')
tf.flush()
# По умолчанию созданный временный файл удаляется, когда файл закрыт
# Передача `delete=False` в NamedTemporaryFile отменяет автоудаление.
Перенаправление потоков ввода и вывода
В Python 3.4+ диспетчер контекста redirect_stdout и redirect_stderr можно использовать для временного перенаправления потоков stdout и stderr.
with open('output.txt', 'w') as f:
with redirect_stdout(f):
print('Hello World!')
redirect_stdout только перенаправляет вызовы stdout из Python, но не из кода библиотеки C.
Чтение и запись в файл inplace
import csv
with inplace(csvfilename, 'rb') as (infh, outfh):
reader = csv.reader(infh)
writer = csv.writer(outfh)
for row in reader:
row += ['new', 'columns']
writer.writerow(row)
Управление пулом процессов
Библиотека multiprocessing Python предоставляет кучу менеджеров контекстов для управления соединениями, пулами и блокировками ресурсов на уровне ОС.
from multiprocessing import Pool
def f(x):
return x*x
with Pool(processes=4) as pool: # запуск 4 worker процесса
# вычисление "f(10)" асинхронно в одномпроцессе
result = pool.apply_async(f, (10,))
print(result.get(timeout=1))
# печатает "100"
print(pool.map(f, range(10)))
# печатает "[0, 1, 4,..., 81]"
Резюме
Короче говоря, Context Managers можно использовать в самых разных случаях. Начните использовать их сразу же, когда заметите шаблон «настройка-завершение», чтобы сделать свой код более питоничным.