5 Декораторов Python, которые я использую почти во всех своих проектах в области Data Science

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

t.me

Python

You will be muted until you read the rules. Read them! @PythonRules A group about the Python programming language. Offtopic things go here: @pythonofftopic Resources to learn python: @pythonres Group for Hy: @hylang Selenium: @SeleniumPython

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

С помощью декораторов мы можем сократить повторение кода и улучшить его читаемость. Я, безусловно, пользуюсь этими инструментами.

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

В проектах по обработке данных и разработке программного обеспечения очень много случаев, когда мы зависим от внешних систем. Не всё всегда находятся под нашим контролем.

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

Я предпочитаю реализовывать эту логику с помощью декоратора retry, который позволяет повторно выполнять программу через N-ное количество времени.

Ниже представлен код, в котором используется данный декоратор:

import time from functools import wraps

def retry(max_tries=3, delay_seconds=1): def decorator_retry(func): @wraps(func) def wrapper_retry(*args, **kwargs): tries = 0 while tries < max_tries: try: return func(*args, **kwargs) except Exception as e: tries += 1 if tries == max_tries: raise e time.sleep(delay_seconds) return wrapper_retry return decorator_retry @retry(max_tries=5, delay_seconds=2) def call_dummy_api(): response = requests.get("https://jsonplaceholder.typicode.com/todos/1") return response

В приведённом выше коде мы пытаемся получить ответ API. Если это не удаётся, мы повторяем одну и ту же задачу 5 раз. Между каждой повторной попыткой мы ждём 2 секунды.

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

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

def memoize(func): cache = {} def wrapper(*args): if args in cache: return cache[args] else: result = func(*args) cache[args] = result return result return wrapper

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

Ниже приведено число Фибоначчи, вычисляющееся функцией. Поскольку это рекуррентная функция, одна и та же вызываемая функция выполняется несколько раз. Но с помощью кэширования мы можем ускорить этот процесс:

@memoize def fibonacci(n): if n <= 1: return n else: return fibonacci(n-1) + fibonacci(n-2)

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

Function slow_fibonacci took 53.05560088157654 seconds to run. Function fast_fibonacci took 7.772445678710938e-05 seconds to run.

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

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

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

Но делать это снова и снова для каждой функции - хлопотно.

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

Вот пример декоратора Python, который выводит время выполнения функции при ее вызове:

import time def timing_decorator(func): def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() print(f"Function {func.__name__} took {end_time - start_time} seconds to run.") return result return wrapper

Вы можете использовать этот декоратор для определения времени выполнения функции:

@timing_decorator def my_function(): # some code here time.sleep(1) # simulate some time-consuming operation return

Вызов функции выведет время, необходимое для запуска.

my_function() >>> Function my_function took 1.0019128322601318 seconds to run.

Этот декоратор в значительной степени является продолжением предыдущего. Но у него есть несколько особых применений.

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

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

Следующий пример иллюстрирует это:

import logging import functools logging.basicConfig(level=logging.INFO) def log_execution(func): @functools.wraps(func) def wrapper(*args, **kwargs): logging.info(f"Executing {func.__name__}") result = func(*args, **kwargs) logging.info(f"Finished executing {func.__name__}") return result return wrapper @log_execution def extract_data(source): # extract data from source data = ... return data @log_execution def transform_data(data): # transform data transformed_data = ... return transformed_data @log_execution def load_data(data, target): # load data into target ... def main(): # extract data data = extract_data(source) # transform data transformed_data = transform_data(data) # load data load_data(transformed_data, target)

Приведённый выше код представляет собой упрощённую версию конвейера ETL. У нас есть три отдельные функции для обработки каждого извлечения, преобразования и загрузки. Мы завернули каждый из них, используя наш декоратор log_execution.

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

INFO:root:Executing extract_data INFO:root:Finished executing extract_data INFO:root:Executing transform_data INFO:root:Finished executing transform_data INFO:root:Executing load_data INFO:root:Finished executing load_data

Мы также могли бы вывести время выполнения внутри этого декоратора. Но я бы хотел, чтобы оба этих функционала были в отдельных декораторах. Таким образом, я могу выбрать, какой из них (или оба) использовать для функции.

Вот как использовать несколько декораторов для одной функции:

@log_execution @timing_decorator def my_function(x, y): time.sleep(1) return x + y

Наконец, очень полезным декоратором в производственных системах является декоратор Notification.

Ещё раз, даже при нескольких повторных попытках хорошо протестированная кодовая база может потерпеть неудачу. И когда это произойдет, нам нужно сообщить кому-нибудь об этом, чтобы принять быстрые меры.

Это не ново, если вы когда-либо создавали конвейер данных и надеялись, что он всегда будет работать без перебоев.

Следующий декоратор отправляет электронное письмо всякий раз, когда выполнение внутренней функции завершается неудачей. В вашем случае это не обязательно должно быть уведомление по электронной почте. Вы можете настроить его для отправки уведомлений Teams / slack:

import smtplib import traceback from email.mime.text import MIMEText def email_on_failure(sender_email, password, recipient_email): def decorator(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: # format the error message and traceback err_msg = f"Error: {str(e)}\n\nTraceback:\n{traceback.format_exc()}" # create the email message message = MIMEText(err_msg) message['Subject'] = f"{func.__name__} failed" message['From'] = sender_email message['To'] = recipient_email # send the email with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp: smtp.login(sender_email, password) smtp.sendmail(sender_email, recipient_email, message.as_string()) # re-raise the exception raise return wrapper return decorator @email_on_failure(sender_email='your_email@gmail.com', password='your_password', recipient_email='recipient_email@gmail.com') def my_function(): # code that might fail

Декораторы - это очень удобный способ добавить новое поведение к нашим функциям. Без них будет много повторений кода.

В этой статье я рассказал о своих наиболее часто используемых декораторах. Вы можете расширить их в соответствии с вашими конкретными потребностями. Например, вы можете использовать сервер Redis для хранения ответов кэша вместо словарей. Это даст вам больше контроля над данными. Или вы могли бы настроить код таким образом, чтобы постепенно увеличивать время ожидания в декораторе Retry.

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

Статья была взята из этого источника:

uproger.com

5 Декораторов Python, которые я использую почти во всех своих проектах в области Data Science

5 Декораторов Python, которые я использую почти во всех своих проектах в области Data Science Поначалу цель каждого разработчика - сделать работающую программу. Постепенно мы начинаем беспокоиться об удобочитаемости и масштабируемости. Именно тогда мы впервые слышим о таком понятии, как декоратор

5 Декораторов Python, которые я использую почти во всех своих проектах в области Data Science

1. Декоратор retry

2. Результаты функции кэширования

3. Функции расчёта времени

4. Логирование вызовов функций

5. Декоратор Notification

Заключение