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

21 января 2022 г. | 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()

Надёжные деструкторы

Контекстные менеджеры дают нам надежный метод очистки ресурсов. В отличие от других языков ООП, таких как 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('https://digitology.tech')) 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')

Настройка моков перед тестированием

with mock.patch('a.b'):
    call()

mock.patch - пример менеджера контекста, который также может использоваться в качестве декоратора. Python 3 поставляется с утилитой ContextDecorator, которая упрощает запись менеджеров контекста.

При использовании в качестве декоратора mock.patch передает вновь созданный мок объект (возвращаемое значение __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('digitology.tech')
...     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]"

Резюме

Короче говоря, менеджеры контекстов можно использовать в самых разных случаях. Начните использовать их сразу же, когда заметите шаблон «настройка-завершение», чтобы сделать свой код более питоничным.