Декоратор для логирования длительных асинхронных функций

Условие

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

import time, asyncio, logging
logging.basicConfig(level=logging.INFO)

def slowlog():
    # ...

@slowlog(limit=2)
async def test_func(wait_time):
    # ...
    time.sleep(wait_time)
    return wait_time

asyncio.run(test_func(1))  # не должно логироваться
asyncio.run(test_func(3))  # должно логироваться

В примере показано, как использовать декоратор: при вызове test_func(1) лог не появляется, а при test_func(3) — появляется, так как время выполнения превышает установленный лимит.

Ответ

Декоратор должен await-ить функцию и сравнивать elapsed с limit. Внутри тестовой корутины замените time.sleep на await asyncio.sleep.

from functools import wraps
from time import perf_counter
import logging
import asyncio

logger = logging.getLogger(__name__)

def slowlog(limit):
    if limit < 0:
        raise ValueError('Порог не может быть отрицательным')
    def decorate(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            started = perf_counter()
            try:
                return await func(*args, **kwargs)
            finally:
                elapsed = perf_counter() - started
                if elapsed > limit:
                    logger.info('%s выполнялась %.3f с', func.__name__, elapsed)
        return wrapper
    return decorate

@slowlog(limit=2)
async def test_func(wait_time):
    await asyncio.sleep(wait_time)
    return wait_time

time.sleep блокировал бы event loop. Здесь логируется и медленное завершение с ошибкой; аргументы и результаты в лог не передаются. Обработчик logging должен быть настроен на INFO.