5 Декораторов Python, которые я использую почти во всех своих проектах в области Data Science
5 Декораторов Python, которые я использую почти во всех своих проектах в области Data Science
Когда мы только учились программировать наша цель, как разработчика была сделать работающую программу. Постепенно мы начинаем беспокоиться об удобочитаемости и масштабируемости. Именно тогда мы впервые слышим о таком понятии, как декораторы.
Декоратор — это функция, которая позволяет обернуть другую функцию для расширения её функциональности без непосредственного изменения её кода.
С помощью декораторов мы можем сократить код и улучшить его читаемость. Я постоянно пользуюсь этими инструментами.
Вот пять наиболее распространённых декораторов, которые я использую почти в каждом проекте с большим объёмом данных.
1. Декоратор retry
В проектах по обработке данных и разработке программного обеспечения очень много случаев, когда мы зависим от внешних систем. Не всё всегда находятся под нашим контролем.
Иногда происходят неожиданное событие, во время которых нам бы хотелось, чтобы внешняя система сама исправляла возникнувшие ошибки и перезапускалась.
Я предпочитаю реализовывать эту логику с помощью декоратора 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 секунды.
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. Декоратор Notification
Наконец, очень полезным декоратором в производственных системах является декоратор 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.
Во всех своих проектах я использую ту или иную версию этих декораторов. Хотя их поведение немного отличается, это общие цели, для которых я часто использую декораторы.