Рубрики
Без рубрики

Магия контекстных менеджеров Python

Управление ресурсами – одна из тех вещей, которые вам нужно сделать на любом языке программирования. Если вы … Tagged with Python, Tuperial, WebDev.

Управление ресурсами – одна из тех вещей, которые вам нужно сделать на любом языке программирования. Независимо от того, имеете ли вы дело с блокировками, файлами, сеансами или подключениями к базе данных – вы всегда должны убедиться, что вы закрываете и освобождаете эти ресурсы для них правильно. Обычно можно делать это с помощью Попробуйте/наконец – Использование ресурса в попробуйте Блок и утилизация этого в Наконец блокировать. В Python, однако, есть лучший способ – Протокол управления контекстом реализовано с использованием с утверждение.

Итак, в этой статье мы рассмотрим, что это такое, как она работает и, самое главное, где вы можете найти и как вы можете реализовать свой собственный удивительный Контекстные менеджеры !

Что такое контекст -менеджер?

Даже если вы не слышали о Python’s Контекст -менеджер Вы уже знаете – на основе вступления – что это замена для Попробуйте/наконец блоки. Он реализован с использованием с оператор обычно используется при открытии файлов. То же, что и с Попробуйте/наконец Этот шаблон был введен, чтобы гарантировать, что некоторая операция будет выполнена в конце блока, даже если произойдет исключение или прекращение программы.

На поверхности Протокол управления контекстом это просто с Заявление, которое окружает блок кода. В действительности он состоит из 2 специальных ( Dunder ) методов – __enter__ и __exit__ – что облегчает настройку и разрыв соответственно.

Когда с Заявление встречается в коде, __enter__ Метод запускается, и его возвратное значение помещается в переменную после как квалификатор. После тела с Блок выполняется, __exit__ Метод вызывается для выполнения разрыва – выполнение роли Наконец блокировать.

# Using try/finally
import time

start = time.perf_counter()  # Setup
try:  # Actual body
    time.sleep(3)
finally:  # Teardown
    end = time.perf_counter()
    elapsed = end - start

print(elapsed)

# Using Context Manager
with Timer() as t:
    time.sleep(3)

print(t.elapsed)

Приведенный выше код показывает оба версии, используя Попробуйте/наконец и более элегантная версия с использованием с оператор для реализации простого таймера. Я упоминал выше, что __enter__ и __exit__ необходимы для реализации такого Контекст -менеджер , но как бы мы их создавали? Давайте посмотрим на код этого Таймер учебный класс:

# Implementation of above context manager
class Timer:
    def __init__(self):
        self._start = None
        self.elapsed = 0.0

    def start(self):
        if self._start is not None:
            raise RuntimeError('Timer already started...')
        self._start = time.perf_counter()

    def stop(self):
        if self._start is None:
            raise RuntimeError('Timer not yet started...')
        end = time.perf_counter()
        self.elapsed += end - self._start
        self._start = None

    def __enter__(self):  # Setup
        self.start()
        return self

    def __exit__(self, *args):  # Teardown
        self.stop()

Этот фрагмент кода показывает Таймер Класс, который реализует оба __enter__ и __exit__ методы __enter__ Метод запускает только таймер и возвращает я который будет назначен в с ... как некоторые_var Анкет После тела с Заявление завершается, __exit__ Метод вызывается с помощью 3 аргументов – тип исключения, значения исключения и Traceback. Если все пойдет хорошо в теле с утверждение, все они будут равны Нет Анкет Если исключение поднимается, они заполнены данными исключений, с которыми мы можем обработать в __exit__ метод В этом случае мы опускаем обработку исключений и просто останавливаем таймер и рассчитываем истеченное время, сохраняя его в атрибуте менеджера контекста.

Мы уже видели здесь как реализацию, так и пример использования с утверждение, но иметь чуть более визуальный пример того, что Действительно Произойдет, давайте посмотрим, как эти специальные методы вызываются без синтаксического сахара Python:

manager = Timer()
manager.__enter__()  # Setup
time.sleep(3)  # Body
manager.__exit__(None, None, None)  # Teardown
print(manager.elapsed)

Теперь, когда мы установили, что такое контекстный менеджер, как он работает и как его реализовать, давайте посмотрим на преимущества его использования – просто чтобы иметь немного больше мотивации, чтобы переключиться с Попробуйте/наконец к с заявления.

Первое преимущество заключается в том, что вся настройка и разрыв происходит под контролем объекта Context Manager. Это предотвращает ошибки и уменьшает код шаблона, что, в свою очередь, делает API более безопасными и проще в использовании. Другая причина его использования – это с Блоки подчеркивают критический раздел и побуждают вас уменьшить объем кода в этом разделе, который также является – в целом – хорошая практика. Наконец – последнее, но не менее важное – это хороший инструмент рефакторинга, который учитывает общую настройку и код разрыва и перемещает его в одно место – __enter__ и __exit__ методы

С учетом сказанного, я надеюсь, что убедил вас начать использовать контекстные менеджеры вместо Попробуйте/наконец Если вы не использовали их раньше. Итак, давайте теперь увидим несколько классных и полезных контекстных менеджеров, которые вы должны начать, включая в своем коде!

Сделать это просто с помощью @ContextManager

В предыдущем разделе мы исследовали, как диспетчер контекста может быть реализован с использованием __enter__ и __exit__ методы Это достаточно просто, но мы можем сделать это еще проще, используя контекст и более конкретно, используя @contextmanager Анкет

@contextmanager это декоратор, который можно использовать для написания автономных функций управления контекстом. Итак, вместо создания всего класса и реализации __enter__ и __exit__ методы , Все, что нам нужно сделать, это создать единый генератор:

from contextlib import contextmanager
from time import time, sleep

@contextmanager
def timed(label):
    start = time()  # Setup - __enter__
    print(f"{label}: Start at {start}")
    try:  
        yield  # yield to body of `with` statement
    finally:  # Teardown - __exit__
        end = time()
        print(f"{label}: End at {end} ({end - start} elapsed)")

with timed("Counter"):
    sleep(3)

# Counter: Start at 1599153092.4826472
# Counter: End at 1599153095.4854734 (3.00282621383667 elapsed)

Этот фрагмент реализует очень похожий контекстный менеджер как Таймер класс в предыдущем разделе. На этот раз нам нужно было гораздо меньше кода. Этот маленький кусочек кода имеет 2 части – все раньше доход И все после доход Анкет Код до доход берет работу __enter__ Метод и доход Сам это возврат утверждение __enter__ метод Все после доход является частью __exit__ метод

Как вы можете видеть выше, создание контекстного диспетчера с использованием одной функции, подобной этой, требует использования Попробуйте/наконец , потому что если исключение происходит в теле с Заявление, это будет поднято на линии с урожай И нам нужно будет справиться с этим в Наконец Блок, который соответствует __exit__ метод

Как я уже упоминал, это можно использовать для автономных контекстных менеджеров. Однако он не подходит для менеджеров контекста, которые должны быть частью объекта, например, соединение или блокировка.

Несмотря на то, что создание контекстного диспетчера с использованием единой функции заставляет вас использовать Попробуйте/наконец И может использоваться только с более простыми вариантами использования, это все еще, на мой взгляд, элегантный и практичный вариант для создания более стройных контекстных менеджеров.

Давайте теперь перейдем от теории к практическим и полезным контекстным менеджерам, которые вы можете построить сами.

Регистрация контекста

Когда придет время попытаться выследить некоторую ошибку в вашем коде, вы, вероятно, сначала посмотрите в журналах, чтобы найти основную причину проблемы. Эти журналы, однако, могут быть установлены по умолчанию для ошибка или Предупреждение Уровень, который может быть недостаточно для отладки целей. Изменение уровня журнала для всей программы должно быть простым, но изменение его для конкретного раздела кода может быть более сложным – это может быть легко решено, хотя с следующим контекстным диспетчера:

import logging
from contextlib import contextmanager

@contextmanager
def log(level):
    logger = logging.getLogger()
    current_level = logger.getEffectiveLevel()
    logger.setLevel(level)
    try:
        yield
    finally:
        logger.setLevel(current_level)

def some_function():
    logging.debug("Some debug level information...")
    logging.error('Serious error...')
    logging.warning('Some warning message...')

with log(logging.DEBUG):
    some_function()

# DEBUG:root:Some debug level information...
# ERROR:root:Serious error...
# WARNING:root:Some warning message...

Тайм -аут контекстного менеджера

В начале этой статьи мы играли с временными блоками кода. Вместо этого мы попробуем установить тайм -ауты на блоки, окруженные с утверждение:

import signal
from time import sleep

class timeout:
    def __init__(self, seconds, *, timeout_message=""):
        self.seconds = int(seconds)
        self.timeout_message = timeout_message

    def _timeout_handler(self, signum, frame):
        raise TimeoutError(self.timeout_message)

    def __enter__(self):
        signal.signal(signal.SIGALRM, self._timeout_handler)  # Set handler for SIGALRM
        signal.alarm(self.seconds)  # start countdown for SIGALRM to be raised

    def __exit__(self, exc_type, exc_val, exc_tb):
        signal.alarm(0)  # Cancel SIGALRM if it's scheduled
        return exc_type is TimeoutError  # Suppress TimeoutError


with timeout(3):
    # Some long running task...
    sleep(10)

Приведенный выше код объявляет класс под названием Тайм -аут Для этого менеджера контекста, поскольку эта задача не может быть выполнена в одной функции. Чтобы иметь возможность реализовать такого рода тайм -аут, нам также необходимо использовать сигналы – более конкретно Сигалрм . Сначала используем Signal.Signal (...) Чтобы установить обработчик на Сигалрм , что означает, что когда Сигалрм поднимается ядром, наша функция обработчика будет вызвана. Что касается этой функции обработчика ( _timeout_handler ), все, что она делает, это повышение Timeouterror , который остановит исполнение в теле с заявление, если это не завершено вовремя. С помощью обработчика мы должны также запустить обратный отсчет с указанного количества секунд, что выполняется Signal.alarm (self.seconds) Анкет

Что касается __exit__ Метод – если тело менеджера контекста удается завершить до истечения времени, Сигалрм будет отменен Signal.alarm (0) и программа может продолжаться. С другой стороны – если сигнал повышается из -за тайм -аута, то _timeout_handler поднимет Timeouterror , который будет пойман и подавлен __exit__ , тело с Заявление будет прервано, а остальная часть кода может продолжать выполнять выполнение.

Помимо контекстных менеджеров выше, уже есть куча полезных в стандартной библиотеке или в других широко используемых библиотеках, таких как запрос или SQLite3 . Итак, давайте посмотрим, что мы можем найти там.

Временно изменить десятичную точность

Если вы выполняете много математических операций и требуете конкретной точности, то вы можете столкнуться с ситуациями, когда вы можете временно изменить точность для десятичных чисел:

from decimal import getcontext, Decimal, setcontext, localcontext, Context

# Bad
old_context = getcontext().copy()
getcontext().prec = 40
print(Decimal(22) / Decimal(7))
setcontext(old_context)

# Good
with localcontext(Context(prec=50)):
    print(Decimal(22) / Decimal(7))  # 3.1428571428571428571428571428571428571428571428571

print(Decimal(22) / Decimal(7))      # 3.142857142857142857142857143

Код выше демонстрирует оба варианта без и с контекстным диспетчера. Второй вариант явно короче и читается. Это также учитывает временный контекст, что делает его менее подверженным ошибкам.

Все вещи из контекста

Мы уже заглянули в контекст При использовании @contextmanager , но есть больше вещей, которые мы можем использовать - в качестве первого примера давайте посмотрим на redirect_stdout и redirect_stderr :

import sys
from contextlib import redirect_stdout

# Bad
with open("help.txt", "w") as file:
    stdout = sys.stdout
    sys.stdout = file
    try:
        help(int)
    finally:
        sys.stdout = stdout

# Good
with open("help.txt", "w") as file:
    with redirect_stdout(file):
        help(int)

Если у вас есть инструмент или функция, которая по умолчанию выводит все в stdout или Stderr , но вы бы предпочли, чтобы они выводили данные где -то еще – например, Для подачи – тогда эти 2 контекстных менеджера могут быть довольно полезными. Как и в предыдущем примере, это значительно улучшает читаемость кода и устраняет ненужный визуальный шум.

Еще один удобный от контекст это подавить Контекст -менеджер, который будет подавлять любые нежелательные исключения и ошибки:

import os
from contextlib import suppress

try:
    os.remove('file.txt')
except FileNotFoundError:
    pass


with suppress(FileNotFoundError):
    os.remove('file.txt')

Определенно предпочтительнее правильно обрабатывать исключения, но иногда вам просто нужно избавиться от этого надоедливого TempercationWarning И этот менеджер контекста, по крайней мере, сделает его читаемым.

Последний от контекст то, что я упомяну, на самом деле мой любимый и это называется Закрытие :

# Bad
try:
    page = urlopen(url)
    ...
finally:
    page.close()

# Good
from contextlib import closing

with closing(urlopen(url)) as page:
    ...

Этот менеджер контекста закроет любой ресурс, переданный ему в качестве аргумента – в случае примера выше – это будет Страница объект. Что касается того, что на самом деле происходит в фоновом режиме – менеджер контекста действительно просто заставляет призывать .close () Метод Страница объект так же, как с Попробуйте/наконец вариант.

Контекстные менеджеры для лучших тестов

Если вы хотите, чтобы люди когда -либо использовали, читали или поддерживаете тест, который вы пишете, вы должны сделать их читаемыми и простыми для понимания и Mock.patch Контекст -менеджер может помочь с этим:

# Bad
import requests
from unittest import mock
from unittest.mock import Mock

r = Mock()
p = mock.patch('requests.get', return_value=r)
mock_func = p.start()
requests.get(...)
# ... do some asserts
p.stop()

# Good
r = Mock()
with mock.patch('requests.get', return_value=r):
    requests.get(...)
    # ... do some asserts

Используя Mock.patch с контекстным менеджером позволяет избавиться от ненужных .Начало () и .stop () вызывает и помогает вам определить четкую область этого конкретного макета. Хорошая вещь об этом в том, что он работает с Unittest а также pytest , хотя это часть стандартной библиотеки (и, следовательно, Unittest ).

Говоря о питест , давайте покажем хотя бы одного очень полезного менеджера контекста из этой библиотеки:

import pytest, os

with pytest.raises(FileNotFoundError, message="Expecting FileNotFoundError"):
    os.remove('file.txt')

Этот пример показывает очень простое использование pytest.raises который утверждает, что блок кода поднимает поставлено исключение. Если это не так, то тест не удается. Это может быть удобно для тестирования путей кода, которые, как ожидается, поднимут исключения или не пройдут.

Устойчивая сессия по запросам

Переходя от pytest в другую великую библиотеку – Запросы Анкет Довольно часто вам может потребоваться сохранить файлы cookie между HTTP -запросами, необходимо сохранить соединение TCP или просто хотеть выполнить несколько запросов на тот же хост. Запросы Предоставляет хороший менеджер контекста, чтобы помочь с этими проблемами, то есть для управления сеансами:

import requests

with requests.Session() as session:
    session.request(method=method, url=url, **kwargs)

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

Управление транзакциями SQLite

И последнее, но не менее важное, есть также менеджер контекста для управления транзакциями SQLite. Помимо устранения вашего кода, этот менеджер контекста также обеспечивает возможность отката от изменений в случае исключения, а также автоматического коммита, если корпус с Заявление завершается успешно:

import sqlite3
from contextlib import closing

# Bad
connection = sqlite3.connect(":memory:")
try:
    connection.execute("INSERT INTO employee(firstname, lastname) values (?, ?)", ("John", "Smith",))
except sqlite3.IntegrityError:
    ...

connection.close()

# Good
with closing(sqlite3.connect(":memory:")) as connection:
    with connection:
        connection.execute("INSERT INTO employee(firstname, lastname) values (?, ?)", ("John", "Smith",))

В этом примере вы также можете увидеть хорошее использование Закрытие Контекст -менеджер, который помогает распоряжаться более не используемым объектом соединения, который еще больше упрощает этот код и гарантирует, что мы не оставляем никаких подключений.

Вывод

Одна вещь, которую я хочу выделить, это то, что контекстные менеджеры-это не просто инструмент для управления ресурсами, а скорее функции, которые позволяют вам извлекать и факторировать общую настройку и разрыв любой пары операций, а не только общие варианты использования, такие как блокировка или сетевые соединения Анкет Это также один из тех великих Pythonic Особенности, которые вы, вероятно, не найдете практически на любом другом языке. Это чисто и элегантно, так что, надеюсь, эта статья показала вам силу контекстных менеджеров и познакомила вас с несколькими способами использования их в вашем коде. 🙂

Оригинал: “https://dev.to/martinheinz/the-magic-of-python-context-managers-5ana”