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

Размещено в категории Python собеседование

18.02.2022 02.05.2022 1166 слов 6 минут

Содержание

Предупреждение

Последний раз данная статья обновлялась 02.05.2022, информация может быть устаревшей.

Список вопросов к Python собеседованию

Краткое резюме по контекстным менеджерам

Контекстные менеджеры нужны там, где есть “настройка” -> блок кода -> “уборка”, при этом “настройку” и “уборку” нужно выполнить в паре.
У контекстного менеджера обязательно присутствуют атрибуты __enter__ и __exit__. Их добавление обеспечивает реализацию протокола контекстного менеджера.
Не обязательно писать целый класс для нового контекстного менеджера, достаточно обернуть генератор в декоратор contextmanager из модуля contextlib.
yield стоит оборачивать в блок try...finally.
Контекстный менеджер определен в PEP 343.
Контекстные менеджеры предназначены для использования в качестве более сжатого механизма управления ресурсами, чем try...finally
Контекстные менеджеры дают нам надежный метод очистки ресурсов (т.к. вызов метода деструктора Python del не всегда гарантируется).

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

Контекстные менеджеры позволяют выделять и освобождать ресурсы именно тогда, когда нам это нужно. Наиболее широко используемым примером контекстных менеджеров является оператор with.

Допустим, у нас есть две связанные операции, которые необходимо выполнить в паре + блок кода между ними. Инструкция with создает контекст выполнения, который позволяет запускать группу операторов под управлением контекстного менеджера.

По сравнению с традиционными конструкциями try ... finally, инструкция with делает код более понятным, безопасным и многоразовым. Многие классы в стандартной библиотеке поддерживают оператор with. Классическим примером этого является open(), который позволяет работать с файловыми объектами используя with:

1
2
with open('some_file', 'w') as opened_file:
    opened_file.write('Hello!')

В общем случае синтаксис использования with выглядит следующим образом:

1
2
with expression as target_var:
    do_something(target_var)

expression, идущее после with должно возвращать объект, реализующий протокол управления контекстом. Этот протокол состоит из двух специальных методов:

__enter__ вызывается оператором with для входа в контекст выполнения;
__exit__ вызывается, когда выполнение покидает блок кода with.

Спецификатор as необязателен. Если мы используем target_var с as, то значение возвращенное методом __enter__ для объекта контекстного менеджера привязывается к этой переменной.

Возврат None

Некоторые контекстные менеджеры возвращают None из __enter__, потому что у них нет объекта, который можно было бы вернуть вызывяющей стороне. В этих случаях использование target_var не имеет смысла.

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

Вызов expression для получения контекстного менеджера.
Сохранение методов контекстного менеджера __enter__ и __exit__ для последующего использования.
Вызов метода __enter__ и сохранение возвращаемого значения в target_var (если target_var используется).
Выполнение блока кода внутри with.
Вызов метода __exit__ в контекстном менеджере после завершения блока кода внутри with.

Наиболее частые сценарии использования:

Open–Close - например, файла, или сокета
Lock–Release - работа с данными в многопоточном приложении
Start–Stop - например, для запуска таймера и его автоматической остановки.
Change–Reset - например, приложение должно подключиться к нескольким источникам данных и у него есть соединение по умолчанию.
Create-Delete
Enter-Exit
Setup-Teardown

Контекстный менеджер из класса

Необходимый минимум функциональности контекстного менеджера требует методов __enter__ и __exit__.

Метод __enter__(self) выполняется до входа в блок. Методу можно возвратить текущий экземпляр класса, что бы к нему можно было обращаться через инструкцию as.
Метод __exit__(self, ex_type, ex_val, ex_trace) выполняется после выхода из блока with, и содержит три параметра — ex_type, ex_value и ex_tr. Переменная ex_type содержит в себе класс исключения, которое было возбуждено, ex_value — сообщение исключения.

Чтобы превратить класс в контекстный менеджер нужно определить в нем два этих метода:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
class FileOpener:
    def __init__(self, f_name, op_type) -> None:
        print("Inside __init__")
        self.file = open(f_name, op_type)

    def __enter__(self):
        print("Inside __enter__")
        return self.file

    def __exit__(self, ex_type, ex_val, ex_trace):
        print("Inside __exit__ %s, %s, %s", ex_type, ex_val, ex_trace)
        self.file.close()
        return True


with FileOpener("test.txt", "w") as file:
    file.write("Test string")
    raise RuntimeError()

Шаги, которые выполняет with при возникновении исключения:

Тип, значение и обратная трассировка ошибки передается в метод __exit__.
Обработка исключения передается методу __exit__
Если __exit__ возвращает True, то исключение было корректно обработано.
При возврате любого другого значения with вызывает исключение.

Контекстный менеджер из генератора

Мы также можем реализовать менеджер контекста через декораторы и генераторы. В Python присутствует модуль contextlib специально для этой цели. Вместо написания класса, мы можем реализовать менеджер контекста из функции-генератора.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
from contextlib import contextmanager

# Общий вид контекстного менеджера
@contextmanager
def some_generator(<arguments>):
    <setup>
    try:
        yield <value>
    finally:
        <cleanup>

# Конкретная реализация
@contextmanager
def open_file(name):
    f = open(name, 'w')
    try:
        yield f
    finally:
        f.close()

with open_file('some_file') as f:
    f.write('Hello!')

Пошаговый разбор данного подхода:

Python встречает ключевое слово yield. Благодаря этому он создает генератор, а не простую функцию.
Благодаря декоратору, contextmanager вызывается с функцией open_file в качестве аргумента.
Функция contextmanager возвращает генератор, обёрнутый в объект GeneratorContextManager.
GeneratorContextManager присваивается функции open_file. Таким образом, когда мы вызовем функцию open_file в следующий раз, то фактически обратимся к объекту GeneratorContextManager.

Контекстный менеджер как декоратор

Можно использовать контекстные менеджеры в качестве декораторов. Для этого при определении класса необходимо наследоваться от класса contextlib.ContextDecorator.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from contextlib import ContextDecorator
from time import time

class RunTime(ContextDecorator):
    """Timing decorator."""

    def __init__(self, description):
        self.description = description

    def __enter__(self):
        print(self.description)
        self.start_time = time()

    def __exit__(self, *args):
        self.end_time = time()
        run_time = self.end_time - self.start_time
        print(f"The function took {run_time} seconds to run.")

@RunTime("This function opens a file")
def custom_file_write(filename, mode, content):
    with open(filename, mode) as f:
        f.write(content)

print(custom_file_write("file.txt", "wt", "Hello"))

Результат:

1
2
3
This function opens a file
The function took 0.0005390644073486328 seconds to run.
None

Вложенные контекстные менеджеры

Инструкция with поддерживает несколько вложенных контекстных менеджеров. Можно использовать любое количество контекстных менеджеров, разделенных запятыми:

1
2
with A() as a, B() as b:
    pass

Асинхронные контекстные менеджеры

Распространенной практикой при написании асинхронных контекстных менеджеров является внедрение четырех специальных методов:

__aenter__
__aexit__
__enter__
__exit__

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# site_checker_v1.py

import aiohttp
import asyncio

class AsyncSession:
    def __init__(self, url):
        self._url = url

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        response = await self.session.get(self._url)
        return response

    async def __aexit__(self, exc_type, exc_value, exc_tb):
        await self.session.close()

async def check(url):
    async with AsyncSession(url) as response:
        print(f"{url}: status -> {response.status}")
        html = await response.text()
        print(f"{url}: type -> {html[:17].strip()}")

async def main():
    await asyncio.gather(
        check("https://realpython.com"),
        check("https://pycoders.com"),
    )

asyncio.run(main())

Примеры менеджеров контекста

Создание сессии в SQLAlchemy

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from contextlib import contextmanager

# an Engine, which the Session will use for connection resources
some_engine = create_engine("sqlite://")

# create a configured "Session" class
Session = sessionmaker(bind=some_engine)


@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

Тайминг выполнения кода

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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

Источники¹²³⁴⁵