Декораторы предоставляют новый и удобный способ для всего, от кэширования до отправки уведомлений.
Во-первых, цель каждого разработчика — заставить все работать. Постепенно мы беспокоимся о читабельности и масштабируемости. Именно тогда мы впервые начинаем думать о декораторах.
Декораторы — отличный способ придать функции дополнительное поведение. И есть небольшие вещи, которые нам, специалистам по данным, часто нужно вводить в определение функции.
С декораторами вы будете удивлены, увидев, насколько вы можете уменьшить повторение кода и улучшить читабельность. Я, конечно, сделал.
Вот пять наиболее распространенных из них, которые я использую почти в каждом проекте с интенсивным использованием данных.
1. Декоратор повторных попыток
В проектах по науке о данных и проектах по разработке программного обеспечения очень много случаев, когда мы зависим от внешних систем. Вещи не находятся под нашим контролем все время.
Когда происходит непредвиденное событие, мы можем захотеть, чтобы наш код подождал некоторое время, позволяя внешней системе исправить себя и перезапуститься.
Я предпочитаю реализовывать эту логику повтора внутри декоратора Python, чтобы я мог аннотировать любую функцию для применения поведения повтора.
Вот код декоратора повторных попыток.
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 секунды.
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.
3. Временные функции
Это не удивительно. При работе с функциями, интенсивно использующими данные, нам не терпится узнать, сколько времени требуется для их выполнения.
Обычный способ сделать это — собрать две метки времени, одну в начале и другую в конце функции. Затем мы можем вычислить продолжительность и распечатать ее вместе с возвращаемыми значениями.
Но делать это снова и снова для нескольких функций — проблема.
Вместо этого мы можем попросить сделать это декоратора. Мы можем аннотировать любую функцию, которая требует вывода длительности.
Вот пример декоратора 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.
4. Регистрация вызовов функций
Этот декоратор во многом является расширением предыдущего декоратора. Но у него есть некоторые особенности использования.
Если вы следуете принципам проектирования программного обеспечения, вы оцените принцип единой ответственности. По сути, это означает, что каждая функция будет иметь свою одну и только одну ответственность.
Когда вы проектируете свой код таким образом, вы также хотите регистрировать информацию о выполнении ваших функций. Вот тут-то и пригодятся декораторы журналирования.
Следующий пример иллюстрирует это.
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
5. Декоратор уведомлений
Наконец, очень полезным декоратором в производственных системах является декоратор уведомлений.
Опять же, даже при нескольких повторных попытках даже хорошо протестированная кодовая база терпит неудачу. И когда это происходит, нам нужно сообщить кому-то об этом, чтобы принять быстрые меры.
Это не ново, если вы когда-либо строили конвейер данных и надеялись, что он будет работать вечно.
Следующий декоратор отправляет электронное письмо всякий раз, когда выполнение внутренней функции завершается сбоем. В вашем случае это не обязательно должно быть уведомление по электронной почте. Вы можете настроить его для отправки уведомлений 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='[email protected]', password='your_password', recipient_email='[email protected]')
def my_function():
# code that might fail
Заключение
Декораторы — это очень удобный способ применить новое поведение к нашим функциям. Без них будет много повторений кода.
В этом посте я обсудил наиболее часто используемые декораторы. Вы можете расширить их для своих конкретных потребностей. Например, вы можете использовать сервер Redis для хранения ответов кеша вместо словарей. Это даст вам больше контроля над данными, такими как постоянство. Или вы можете настроить код, чтобы постепенно увеличивать время ожидания в декораторе повторных попыток.
Во всех своих проектах я использую некоторые версии этих декораторов. Хотя их поведение немного отличается, это общие цели, для которых я часто использую декораторы.
Я надеюсь, что этот пост поможет вам.
Спасибо за прочтение, друг! Если вам понравилась моя статья, давайте поддерживать связь в LinkedIn, Twitter и Medium.
Еще не являетесь участником Medium? Пожалуйста, используйте эту ссылку, чтобы стать участником, потому что без каких-либо дополнительных затрат для вас я получаю небольшую комиссию за рекомендацию вас.
