Как в Python создать декоратор для асинхронной функции?

Ответ

Чтобы создать декоратор для асинхронной (async def) функции, его внутренняя функция-обертка (wrapper) также должна быть асинхронной и использовать await для вызова декорируемой функции.

Важно использовать @functools.wraps для сохранения метаданных исходной функции (имени, документации), что упрощает отладку.

Основной шаблон:

import asyncio
import functools
import time

def async_timer_decorator(func):
    """Декоратор для измерения времени выполнения асинхронной функции."""
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        # Обертка должна быть async def
        print(f"Вызов функции {func.__name__}...")
        start_time = time.time()

        # Вызов оригинальной функции через await
        result = await func(*args, **kwargs)

        end_time = time.time()
        print(f"Функция {func.__name__} выполнилась за {end_time - start_time:.4f} секунд")
        return result
    return wrapper

@async_timer_decorator
async def some_async_task():
    print("Выполняется асинхронная задача...")
    await asyncio.sleep(1)
    return "Задача завершена"

Ключевые принципы:

  • Асинхронная обертка: Внутренняя функция wrapper объявляется с async def.
  • Использование await: Оригинальная функция вызывается с помощью await func(...).
  • Сохранение метаданных: Декоратор @functools.wraps(func) применяется к wrapper для корректного наследования атрибутов (__name__, __doc__ и т.д.).

Пример использования:

async def main():
    result = await some_async_task()
    print(result)

# Запуск в асинхронном контексте
asyncio.run(main())

# Вывод:
# Вызов функции some_async_task...
# Выполняется асинхронная задача...
# Функция some_async_task выполнилась за 1.0012 секунд
# Задача завершена

Ответ 18+ 🔞

А, ну ты глянь, какой вопрос подъехал! Декораторы для асинхронных функций, да? Ну это ж, блядь, классика жанра, как Герасим с Муму, только в мире питона. Сейчас разжуём, сука, до последней косточки.

Так, слушай сюда, распиздяй. Главная засада, на которую все наступают — это забыть, что если функция async def, то и обёртка вокруг неё должна быть такой же, асинхронной тварью. И вызывать её надо не просто func(), а через await, иначе получишь не результат, а какую-то хуйню — корутину, которая так и будет висеть, как недоделанная мысль.

И ещё, ёпта, важный момент — не забудь про @functools.wraps. Это чтобы твой декоратор не сожрал имя и документацию оригинальной функции. Без этого при отладке будешь как тот немой Герасим — только «Му-му» мычать, а понять, какая функция где сломалась, нихуя не получится.

Вот смотри, как это выглядит в коде, чтоб ты не обосрался с первого раза:

import asyncio
import functools
import time

def async_timer_decorator(func):
    """Декоратор для измерения времени выполнения асинхронной функции."""
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        # Обертка должна быть async def
        print(f"Вызов функции {func.__name__}...")
        start_time = time.time()

        # Вызов оригинальной функции через await
        result = await func(*args, **kwargs)

        end_time = time.time()
        print(f"Функция {func.__name__} выполнилась за {end_time - start_time:.4f} секунд")
        return result
    return wrapper

@async_timer_decorator
async def some_async_task():
    print("Выполняется асинхронная задача...")
    await asyncio.sleep(1)
    return "Задача завершена"

Видишь? Всё просто, как три копейки. Обёртка wrapperasync def. Вызов функции — через await. И @functools.wraps(func) сверху, чтобы не потерять лицо, блядь, в глазах интерпретатора.

А запускается это вот так, смотри:

async def main():
    result = await some_async_task()
    print(result)

# Запуск в асинхронном контексте
asyncio.run(main())

# Вывод:
# Вызов функции some_async_task...
# Выполняется асинхронная задача...
# Функция some_async_task выполнилась за 1.0012 секунд
# Задача завершена

Вот и вся магия, сука. Главное — не перепутай синхронное с асинхронным, а то будет как в том анекдоте: «Хотел как лучше, а получил пизда».