Декораторы в Python: как писать обёртки с аргументами, сохранять сигнатуру и работать с async

Декоратор в Python — это инструмент модификации поведения функций без прямого изменения их исходного кода. На практике базовая конструкция def wrapper(*args, **kwargs) быстро упирается в ограничения: затираются имена функций и документация, ломается генерация схем в веб-фреймворках (например, FastAPI), а асинхронные корутины перестают корректно обрабатываться.

Разберём, как устроены декораторы на уровне замыканий, как сохранять сигнатуры и метаданные с помощью functools.wraps, как передавать параметры в обёртки и как корректно обрабатывать асинхронный код.

---

Как устроен декоратор под капотом и зачем нужен functools.wraps

Синтаксис декоратора @decorator представляет собой синтаксический сахар для вызова функции высшего порядка:

@decorator
def target():
    pass

# Полный эквивалент:
# target = decorator(target)

Декоратор принимает исходную функцию, создаёт новую функцию-обёртку (wrapper) и возвращает её вместо оригинала.

Проблема затирания метаданных

Если написать обёртку вручную без вспомогательных инструментов, возвращённый объект заменит исходные метаданные своими:

def naive_decorator(func):
    def wrapper(*args, **kwargs):
        """Документация обёртки."""
        return func(*args, **kwargs)
    return wrapper

@naive_decorator
def calculate_tax(amount: float) -> float:
    """Вычисляет налог на сумму."""
    return amount * 0.2

print(calculate_tax.__name__)  # 'wrapper'
print(calculate_tax.__doc__)   # 'Документация обёртки.'

В результате системы логирования, профилировщики, автодокументация (Sphinx) и генераторы OpenAPI-схем теряют исходное имя calculate_tax, его docstring и аннотации типов.

Роль functools.wraps и атрибут __wrapped__

Для решения этой проблемы стандартная библиотека предоставляет декоратор functools.wraps (построенный на базе functools.update_wrapper). Он копирует ключевые метаданные исходной функции (__module__, __name__, __qualname__, __doc__, __annotations__), обновляет __dict__ и добавляет специальный атрибут __wrapped__, указывающий на оригинальный объект:

from functools import wraps

def proper_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@proper_decorator
def calculate_tax(amount: float) -> float:
    """Вычисляет налог на сумму."""
    return amount * 0.2

print(calculate_tax.__name__)  # 'calculate_tax'
print(calculate_tax.__doc__)   # 'Вычисляет налог на сумму.'

Как inspect.signature() читает декорированную функцию

Модуль inspect использует атрибут __wrapped__ для интроспекции. Функция inspect.signature() автоматически переходит по цепочке __wrapped__ до исходной функции и возвращает её реальные параметры, а не обобщённые (*args, **kwargs):

from inspect import signature, unwrap

sig = signature(calculate_tax)
print(sig)  # (amount: float) -> float

# Доступ к исходной функции без обёртки:
original_func = unwrap(calculate_tax)
print(original_func is calculate_tax.__wrapped__)  # True

---

Создание декоратора с аргументами (фабрика декораторов)

Если декоратору требуются собственные параметры конфигурации (таймаут, количество повторов, префикс логов), используется трёхслойная структура — фабрика декораторов.

Трёхслойная структура замыкания

1. Внешний слой (фабрика): принимает настройки декоратора.
2. Средний слой (декоратор): принимает декорируемую функцию.
3. Внутренний слой (обёртка): принимает аргументы целевой функции (*args, **kwargs).

from functools import wraps
from typing import Callable, Any

def retry(times: int, exception_cls: type[BaseException] = Exception):
    """Фабрика декоратора: принимает параметры конфигурации."""
    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        """Собственно декоратор: принимает функцию."""
        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            """Обёртка: принимает аргументы вызова функции."""
            last_exc = None
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exception_cls as exc:
                    last_exc = exc
            raise RuntimeError(f"Операция провалена после {times} попыток") from last_exc
        return wrapper
    return decorator

Применение:

@retry(times=3, exception_cls=ValueError)
def parse_payload(data: dict) -> str:
    if not data.get("valid"):
        raise ValueError("Некорректные данные")
    return data["payload"]

Механика работы: сначала вызывается retry(times=3, ...), которая возвращает объект decorator. Затем этот объект вызывается с аргументом parse_payload: parse_payload = retry(times=3)(parse_payload).

---

Порядок выполнения нескольких декораторов

При использовании нескольких декораторов важно разделять:

• Время инициализации (декорирования): срабатывает снизу вверх.
• Время выполнения (вызова): срабатывает сверху вниз (снаружи внутрь).

@decorator_one
@decorator_two
def process():
    pass

# Эквивалентно:
# process = decorator_one(decorator_two(process))

Схема прохождения вызова через цепочку обёрток:

Входящий вызов process()
       │
       ▼
┌───────────────────────────────┐
│ decorator_one (внешний)       │
│   ┌─────────────────────────┐ │
│   │ decorator_two           │ │
│   │   ┌───────────────────┐ │ │
│   │   │ Исходная функция  │ │ │
│   │   └───────────────────┘ │ │
│   │ decorator_two (выход)   │ │
│   └─────────────────────────┘ │
│ decorator_one (выход)         │
└───────────────────────────────┘
       │
       ▼
Возврат результата

Пример с логами вызовов:

from functools import wraps

def wrap_a(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print("Вход в A")
        res = func(*args, **kwargs)
        print("Выход из A")
        return res
    return wrapper

def wrap_b(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print("Вход в B")
        res = func(*args, **kwargs)
        print("Выход из B")
        return res
    return wrapper

@wrap_a
@wrap_b
def work():
    print("Выполнение work")

work()

Вывод в консоли:

Вход в A
Вход в B
Выполнение work
Выход из B
Выход из A

---

Поддержка асинхронных функций (async / await)

Синхронный декоратор не может корректно замерить или модифицировать асинхронную функцию. Если синхронный wrapper вызывает func(*args, **kwargs), где func объявлена через async def, вызов вернёт не результат, а объект корутины (coroutine object), который ещё не выполнился.

Специализированный async-декоратор

Для асинхронных функций сама обёртка должна быть объявлена как async def и содержать await:

import asyncio
from functools import wraps
from typing import Callable, Any

def async_logger(func: Callable[..., Any]) -> Callable[..., Any]:
    @wraps(func)
    async def wrapper(*args: Any, **kwargs: Any) -> Any:
        print(f"[ASYNC START] {func.__name__}")
        result = await func(*args, **kwargs)
        print(f"[ASYNC END] {func.__name__}")
        return result
    return wrapper

@async_logger
async def fetch_remote_data(endpoint: str) -> dict:
    await asyncio.sleep(0.05)
    return {"status": 200, "endpoint": endpoint}

Универсальный декоратор через inspect.iscoroutinefunction

Чтобы декоратор корректно работал как с синхронными, так и с асинхронными функциями, определяют тип вызываемого объекта во время декорирования с помощью inspect.iscoroutinefunction:

import inspect
from functools import wraps
from typing import Callable, Any

def universal_tracer(func: Callable[..., Any]) -> Callable[..., Any]:
    if inspect.iscoroutinefunction(func):
        @wraps(func)
        async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
            print(f"Async call: {func.__name__}")
            return await func(*args, **kwargs)
        return async_wrapper
    else:
        @wraps(func)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            print(f"Sync call: {func.__name__}")
            return func(*args, **kwargs)
        return sync_wrapper

---

Практический проект: Декоратор замера времени выполнения (timed) и тесты

Реализуем параметризованный декоратор timed, который:

1. Принимает строковый ярлык (label), единицу измерения (unit: "s" или "ms") и функцию вывода (sink).
2. Поддерживает как синхронные, так и асинхронные функции.
3. Гарантированно замеряет время с помощью time.perf_counter() даже при возникновении исключений (try...finally).
4. Сохраняет метаданные и сигнатуру.

Реализация

import asyncio
import inspect
import time
from functools import wraps
from typing import Callable, Any, Literal

def timed(
    label: str | None = None,
    unit: Literal["s", "ms"] = "ms",
    sink: Callable[[str], None] = print
):
    """Декоратор для замера времени выполнения синхронных и асинхронных функций."""
    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        display_name = label or func.__qualname__

        def format_message(duration_sec: float) -> str:
            if unit == "ms":
                val = duration_sec * 1000
                return f"[{display_name}] выполнено за {val:.3f} ms"
            return f"[{display_name}] выполнено за {duration_sec:.6f} s"

        if inspect.iscoroutinefunction(func):
            @wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                start = time.perf_counter()
                try:
                    return await func(*args, **kwargs)
                finally:
                    elapsed = time.perf_counter() - start
                    sink(format_message(elapsed))
            return async_wrapper

        @wraps(func)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                elapsed = time.perf_counter() - start
                sink(format_message(elapsed))
        return sync_wrapper

    return decorator

Проверочные тесты и assertions

Проверим корректность работы декоратора для синхронного и асинхронного сценариев, а также сохранение сигнатуры:

import asyncio
from inspect import signature

# Хранилище для перехваченных сообщений в тестах
logs: list[str] = []

def test_sink(msg: str) -> None:
    logs.append(msg)

@timed(label="sync_calc", unit="ms", sink=test_sink)
def compute_sum(a: int, b: int = 10) -> int:
    """Вычисляет сумму чисел."""
    time.sleep(0.01)
    return a + b

@timed(label="async_fetch", unit="ms", sink=test_sink)
async def async_get_data(item_id: int) -> str:
    """Асинхронно получает элемент."""
    await asyncio.sleep(0.01)
    return f"item_{item_id}"

# 1. Проверка сохранения метаданных и сигнатуры
assert compute_sum.__name__ == "compute_sum"
assert compute_sum.__doc__ == "Вычисляет сумму чисел."
assert str(signature(compute_sum)) == "(a: int, b: int = 10) -> int"

assert async_get_data.__name__ == "async_get_data"
assert async_get_data.__doc__ == "Асинхронно получает элемент."
assert str(signature(async_get_data)) == "(item_id: int) -> str"

# 2. Проверка работы синхронной функции
res_sync = compute_sum(5, 15)
assert res_sync == 20
assert len(logs) == 1
assert "[sync_calc] выполнено за" in logs[0]

# 3. Проверка работы асинхронной функции
async def run_async_test():
    res_async = await async_get_data(42)
    assert res_async == "item_42"
    assert len(logs) == 2
    assert "[async_fetch] выполнено за" in logs[1]

asyncio.run(run_async_test())
print("Все тесты успешно пройдены!")

---

Частые ошибки и Best Practices

1. Забытый return внутри wrapper.
   Если в теле обёртки вызвать func(*args, **kwargs), но не вернуть результат через return, функция всегда будет возвращать None.

2. Забытые скобки при вызове параметризованного декоратора.
   Если декоратор написан как фабрика (принимает аргументы), его обязательно вызывать со скобками @my_decorator(), даже если все параметры имеют значения по умолчанию. Иначе первым аргументом фабрики станет сама функция, и логика декоратора сломается.

3. Игнорирование блоков try...finally.
   Любые действия пост-обработки (замер времени, закрытие соединений, снятие блокировок, сбор метрик) должны находиться в блоке finally. В противном случае при выбросе исключения внутри целевой функции логика завершения не выполнится.

4. Мутация изменяемых аргументов.
   Если декоратор читает *args или **kwargs, избегайте их непреднамеренной модификации (например, вызова .pop()), если это не является прямой задачей декоратора.

---

FAQ (Часто задаваемые вопросы)

1. Зачем использовать functools.wraps, если код работает и без него?

Без @wraps(func) функция теряет исходное имя (__name__), документацию (__doc__) и аннотации типов, заменяя их данными внутренней функции wrapper. Это ломает работу отладчиков, автогенерацию документации и валидацию в веб-фреймворках (например, внедрение зависимостей в FastAPI).

2. Как передать аргументы и в декоратор, и в саму декорируемую функцию?

Используется трёхслойное замыкание:

• Внешняя функция принимает параметры настройки декоратора.
• Средняя функция принимает целевую функцию.
• Внутренний wrapper(*args, **kwargs) принимает аргументы целевой функции.

3. В каком порядке выполняются несколько декораторов?

При объявлении они применяются снизу вверх (внутренний декоратор оборачивает функцию первым). При вызове функции код выполняется сверху вниз: управление сначала попадает в верхний декоратор, затем во вложенные, и в конце — в целевую функцию.

4. Можно ли написать один декоратор, работающий и с sync, и с async?

Да. Для этого внутри декоратора проверяется тип функции с помощью inspect.iscoroutinefunction(func). Если функция асинхронная, возвращается async def wrapper, в противном случае — синхронный def wrapper.

5. Как получить доступ к исходной функции без наложенных декораторов?

Если декораторы использовали functools.wraps, исходная функция доступна через атрибут func.__wrapped__. Чтобы снять сразу несколько слоёв обёрток до самого основания, используется функция inspect.unwrap(func).

---

Вывод

Грамотная разработка декораторов в Python строится на трёх правилах:

• Использование functools.wraps обязательно для сохранения метаданных и поддержки inspect.signature().
• Трёхслойная структура фабрики декораторов позволяет гибко передавать параметры конфигурации.
• Проверка через inspect.iscoroutinefunction обеспечивает корректную обработку асинхронных корутин без блокировки event loop.

Источники

• functools — Higher-order functions and operations on callable ...
• 8. Compound statements — Python 3.15.0 documentation
• Inspect live objects
• Command-line introspection tools
• Python Development Mode
• Coroutines and tasks