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

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

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

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

Допустим, у нас есть две связанные операции, которые необходимо выполнить в паре + блок кода между ними. Инструкция 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 должно возвращать объект, реализующий протокол управления контекстом. Этот протокол состоит из двух специальных методов:

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

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

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

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

  1. Вызов expression для получения контекстного менеджера.
  2. Сохранение методов контекстного менеджера __enter__ и __exit__ для последующего использования.
  3. Вызов метода __enter__ и сохранение возвращаемого значения в target_var (если target_var используется).
  4. Выполнение блока кода внутри with.
  5. Вызов метода __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 при возникновении исключения:

  1. Тип, значение и обратная трассировка ошибки передается в метод __exit__.
  2. Обработка исключения передается методу __exit__
  3. Если __exit__ возвращает True, то исключение было корректно обработано.
  4. При возврате любого другого значения 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!')

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

  1. Python встречает ключевое слово yield. Благодаря этому он создает генератор, а не простую функцию.
  2. Благодаря декоратору, contextmanager вызывается с функцией open_file в качестве аргумента.
  3. Функция contextmanager возвращает генератор, обёрнутый в объект GeneratorContextManager.
  4. 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

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

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

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

  1. __aenter__
  2. __aexit__
  3. __enter__
  4. __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())
 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

Источники12345