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

Оболочки Python

Обертки Python позволяют добавить новую функциональность или модифицировать ее поведение без непосредственного изменения исходного кода. 

Обертки можно использовать в различных сценариях:

Расширение функциональности: Мы можем задействовать декоратор и добавить такие функции, как ведение журнала, измерение производительности или кэширование.

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

Модификация поведения: Мы можем, к примеру, проверять аргументы функции, не используя многочисленные строки assert.

Примеры:

1 — Timer

Эта функция измеряет время выполнения операции и выводит прошедшее время. Её можно использовать  для анализа кода и его оптимизации .

import time

def timer(func):
    def wrapper(*args, **kwargs):
        # start the timer
        start_time = time.time()
        # call the decorated function
        result = func(*args, **kwargs)
        # remeasure the time
        end_time = time.time()
        # compute the elapsed time and print it
        execution_time = end_time - start_time
        print(f"Execution time: {execution_time} seconds")
        # return the result of the decorated function execution
        return result
    # return reference to the wrapper function
    return wrapper

Для создания декоратора в Python нам необходимо определить функцию timer, которая получает параметр func. Это указывает на то, что это функция-декоратор. Внутри функции timer мы определяем другую функцию — wrapper. wrapper — это обёртка, которая обычно принимает аргументы декорируемой функции.

Внутри функции-обертки мы запускаем нужную функцию, используя указанные аргументы. Это можно сделать с помощью строки: result = func(*args, **kwargs).

Функция-декоратор возвращает запрос на функцию-обертку, которую мы только что создали.

Чтобы использовать декоратор, можно применить его к нужной функции с помощью символа @.

@timer
def train_model():
    print("Starting the model training function...")
    # simulate a function execution by pausing the program for 5 seconds
    time.sleep(5) 
    print("Model training completed!")

train_model() 

Что получим в итоге:

Starting the model training function…

Model Training completed!

Execution time: 5.006425619125366 seconds

2 — debug

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

def debug(func):
    def wrapper(*args, **kwargs):
        # print the fucntion name and arguments
        print(f"Calling {func.__name__} with args: {args} kwargs: {kwargs}")
        # call the function
        result = func(*args, **kwargs)
        # print the results
        print(f"{func.__name__} returned: {result}")
        return result
    return wrapper

Мы можем использовать параметры __name__ для того, чтобы получить имя вызываемой функции, а затем параметры args, kwargs для вывода того, что было предано функции.

@debug
def add_numbers(x, y):
    return x + y
add_numbers(7, y=5,)  # Output: Calling add_numbers with args: (7) kwargs: {'y': 5} \n add_numbers returned: 12

3 — Exception Handler

Функция exception_handler будет ловить любые исключения, возникающие в функции-обертке, и обрабатывать их в зависимости от ситуации.

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

def exception_handler(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # Handle the exception
            print(f"An exception occurred: {str(e)}")
            # Optionally, perform additional error handling or logging
            # Reraise the exception if needed
    return wrapper

Это помогает навести порядок в коде и установить единую процедуру обработки исключений и протоколирования ошибок.

@exception_handler
def divide(x, y):
    result = x / y
    return result
divide(10, 0)  # Output: An exception occurred: division by zero

4 — Input Validator

Эта функция-обертка проверяет исходные аргументы функции на соответствие заданным условиям или типам данных. Она может быть использована для обеспечения корректности и согласованности входных данных.

Чтобы добавить валидацию в декоратор, необходимо «обернуть» функцию декоратора другой функцией, принимающей в качестве аргументов одну или несколько функций валидации. Эти функции проверки отвечают за проверку соответствия вводимых значений определенным критериям или условиям.

Сама функция validate_input теперь выступает в роли декоратора. Внутри функции-обертки входные данные и аргументы ключевых слов проверяются на соответствие предоставленным функциям проверки. Если какой-либо аргумент не проходит проверку, то возникает ошибка ValueError с сообщением о недопустимости аргумента.

def validate_input(*validations):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for i, val in enumerate(args):
                if i < len(validations):
                    if not validations[i](val):
                        raise ValueError(f"Invalid argument: {val}")
            for key, val in kwargs.items():
                if key in validations[len(args):]:
                    if not validations[len(args):][key](val):
                        raise ValueError(f"Invalid argument: {key}={val}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

Для запуска валидированного ввода необходимо определить функции валидации. Например, можно использовать две функции проверки. Первая функция (lambda x: x > 0) проверяет, что аргумент x больше 0, а вторая функция (lambda y: isinstance(y, str)) проверяет, что аргумент y имеет тип string.

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

@validate_input(lambda x: x > 0, lambda y: isinstance(y, str))
def divide_and_print(x, message):
    print(message)
    return 1 / x

divide_and_print(5, "Hello!")  # Output: Hello! 1.0

5 — Retry

Эта обёртка позволяет повторить запуск функции заданное количество раз с задержкой между повторами — удобно при работе с сетевыми или API-вызовами, которые не выполняются из-за каких-либо проблем.

Для запуска функции мы можем определить еще одну функцию-обертку для нашего декоратора, аналогично предыдущему примеру. Однако на этот раз вместо того, чтобы предоставлять функции валидации в качестве входных переменных, мы можем передать конкретные параметры — max_attemps и delay.

При запуске декорированной функции вызывается функция-обертка. Она отслеживает количество попыток (начиная с 0) и переходит в цикл while. Цикл пытается выполнить декорированную функцию и в случае успеха немедленно посылает результат. Если же произошло какое-то отклонение, то цикл увеличивает счетчик попыток и выводит сообщение об ошибке с указанием номера попытки и конкретного отклонения. Затем цикл ждет заданную задержку с помощью функции time.sleep, после чего повторяет попытку выполнения функции.

import time

def retry(max_attempts, delay=1):
    def decorator(func):
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    print(f"Attempt {attempts} failed: {e}")
                    time.sleep(delay)
            print(f"Function failed after {max_attempts} attempts")
        return wrapper
    return decorator

Для вызова функции можно указать максимальное количество попыток и продолжительность времени в секундах между каждым вызовом функции.

@retry(max_attempts=3, delay=2)
def fetch_data(url):
    print("Fetching the data..")
    # raise timeout error to simulate a server not responding..
    raise TimeoutError("Server is not responding.")
fetch_data("https://example.com/data")  # Retries 3 times with a 2-second delay between attempts
От редакции

28 августа начнется новый поток по языку программирования Python. На нем мы разберем: библиотеки Python и решение конкретных задач DevOps, правила эффективного и поддерживаемого кода, принципы автоматизации: Docker, Gitlab, Prometheus, K8S и многое другое.

Узнать больше о потоке вы можете на нашем сайте: ссылка

Комментарии (8)


  1. AlexTeexone
    17.07.2023 07:39
    +1

    Кроме декораторов еще очень удобно использовать профайлинг в виде контекстных менеджеров: с ними можно запускать код внутри with выражения. Увидел такое впервые в коде YOLOv8, с тех пор пользуюсь

    Ccылка на Github


  1. NewSouth
    17.07.2023 07:39
    +8

    1. exception_handler  - скорее exception_silencer. Очень плохо использовать такие декораторы и в целом конструкции вида:

      try:             
        return func(*args, **kwargs)
      except Exception as e:
        print(f"An exception occurred: {str(e)}")

    при действительной ошибке внутри функции она будет заглушена, никто о ней не узнает - а workflow пойдёт дальше так, как будто функция вернула None - и вся программа поведёт себя непредсказуемо.

    1. validate_input - похоже на какую-то попытку сделать из нетипизированного языка типизированный. Можно подумать, как будто проверять каждый раз тип аргумента - lambda y: isinstance(y, str) - хорошая идея, но это совсем не так. Нужен типизированный язык - есть полно таких языков, не нужно переделывать Python.

    2. timer - давно уже рекомендуется вместо time.time() использовать time.perf_counter()

      Я лично пользовался таким декоратором: соединенные вместе декораторы timer и debug + счетчик отступов:

    def performance_debug(f):
        performance_debug.active = 0
    
        def tt(self, *args, **kwargs):
            turn_on = getattr(self, 'performance_debug', False)
            if turn_on:
                performance_debug.active += 1
                t0 = time.perf_counter()
                tabs = '\t'*(performance_debug.active - 1)
                name = f.__name__
                print('{tabs}Executing <{name}>'.format(tabs=tabs, name=name))
                res = f(self, *args, **kwargs)
                print('{tabs}Function <{name}> execution time: {time:.3f} ms'.format(
                    tabs=tabs, name=name, time=(time.perf_counter() - t0)*1000))
                performance_debug.active -= 1
                return res
            else:
                return f(self, *args, **kwargs)
        return tt


    1. uuger
      17.07.2023 07:39
      +5

      exception_handler  - скорее exception_silencer. Очень плохо использовать такие декораторы и в целом конструкции вида

      Натыкался на похожий момент в работе (язык другой был, но суть та же) - последствия были катастрофические: валящийся в случайные моменты сервис при кристально чистых логах, а код был от подрядчика, который сдал работы и занялся другими проектами. В итоге, расследование было долгим, нудным и печальным, а главное - дорогим в контексте стоимости простоя сервиса. Люди, пожалуйста, не делайте так


    1. Tangeman
      17.07.2023 07:39
      +2

      validate_input - похоже на какую-то попытку сделать из нетипизированного языка типизированный.

      Это похоже на попытку убедиться что на входе получено ровно то что ожидается - очень хорошая практика.

      Как минимум валидация помогает при отладке, как максимум - во всех остальных случаях когда у вас нет 101% уверенности что на входе будут только ожидаемые данные, например, если вы пишите библиотеку, или если значения аргументов получены извне (введены пользователем, переданы API etc).

      Гораздо проще аннотировать аргументы и отдать валидацию декоратору чем вручную это делать внутри каждой функции где существует риск, не говоря уже про DRY.

      Дополнительная польза валидации на входе - это чёткое сообщение о том что и где пошло не так, а не непонятное исключение где-то в глубине которое возникло по причине неверного аргумента.

      Впрочем, расчитывать на то что неправильный аргумент обязательно вызовет исключение не стоит - к примеру, если у вас на входе ожидается целое число в диапазоне от 3 до 7 и оно используется в вычислениях, но при этом на вход кто-то подал 13 - исключения не будет, будет неверный результат.

      В конце концов, pydantic и его @validate_arguments не просто так появились - тот кто игнорирует валидацию обязательно нарвётся, или хуже того - подставит кого-то ещё.


  1. burdin
    17.07.2023 07:39
    +2

    Для валидации я сделал вот такой декоратор:

    https://github.com/EvgeniyBurdin/valdec

    А для исключений (ну и заодно, для обработки "до" и "после"), вот такой:

    https://github.com/EvgeniyBurdin/exdec


  1. Skpd
    17.07.2023 07:39
    +3

    Вместо retry - tenacity.


  1. hippowdon
    17.07.2023 07:39
    +1

    Если вам нужен retry - рекомендую библиотеку tenacity https://pypi.org/project/tenacity/

    Она предлагает retry декоратор с разными политиками.


  1. Jolt
    17.07.2023 07:39

    Ааа-ааа! Уберите от экранов джунов, они потом такое начнут в реальный код вставлять!