Для каких целей писал декораторы

Применение декораторов в разработке

Декораторы — один из мощнейших инструментов Python. За 10+ лет я использовал их для решения множества практических задач, от кеширования до контроля доступа.

Логирование и мониторинг

Одно из главных применений — логирование вызовов функций и профилирование производительности.

import functools
import logging
import time

logger = logging.getLogger(__name__)

def log_function_call(func):
    """Логирует все вызовы функции с параметрами и временем выполнения"""
    @functools.wraps(func)
    async def async_wrapper(*args, **kwargs):
        start = time.time()
        logger.info(f"Calling {func.__name__} with args={args}, kwargs={kwargs}")
        try:
            result = await func(*args, **kwargs)
            duration = time.time() - start
            logger.info(f"{func.__name__} completed in {duration:.2f}s")
            return result
        except Exception as e:
            logger.error(f"{func.__name__} failed: {e}", exc_info=True)
            raise
    return async_wrapper

@log_function_call
async def process_order(order_id: int):
    await asyncio.sleep(1)
    return {"status": "success"}

Кеширование результатов

Для оптимизации дорогостоящих операций использую кеширование с TTL.

from datetime import datetime, timedelta
from typing import Any, Callable

def cached(ttl_seconds: int = 300):
    """Кеширует результат функции на указанное время"""
    def decorator(func: Callable) -> Callable:
        cache = {}
        cache_time = {}
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            now = datetime.now()
            
            if key in cache and now - cache_time[key] < timedelta(seconds=ttl_seconds):
                return cache[key]
            
            result = func(*args, **kwargs)
            cache[key] = result
            cache_time[key] = now
            return result
        
        return wrapper
    return decorator

@cached(ttl_seconds=600)
def get_user_profile(user_id: int):
    # Дорогая операция БД
    return db.query(User).filter(User.id == user_id).first()

Аутентификация и авторизация

Декораторы идеальны для проверки прав доступа в FastAPI и Flask.

from fastapi import Depends, HTTPException, status
from functools import wraps

def require_role(required_role: str):
    """Проверяет, что пользователь имеет нужную роль"""
    async def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, current_user: User = Depends(get_current_user), **kwargs):
            if current_user.role != required_role:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail="Access denied"
                )
            return await func(*args, **kwargs)
        return wrapper
    return decorator

@app.get("/admin/users")
@require_role("admin")
async def list_users():
    return await User.get_all()

Валидация параметров

Декораторы помогают валидировать входные данные до вызова основной логики.

def validate_email(func):
    """Проверяет, что email параметр валиден"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        email = kwargs.get('email')
        if email and '@' not in email:
            raise ValueError(f"Invalid email: {email}")
        return func(*args, **kwargs)
    return wrapper

@validate_email
def send_notification(user_id: int, email: str, message: str):
    # Отправляем уведомление
    pass

Retry логика

Для работы с внешними API часто нужны повторные попытки при ошибках.

import asyncio

def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
    """Повторяет функцию при ошибке с экспоненциальной задержкой"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            attempt = 0
            current_delay = delay
            
            while attempt < max_attempts:
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    attempt += 1
                    if attempt >= max_attempts:
                        raise
                    logger.warning(f"Attempt {attempt} failed, retrying in {current_delay}s: {e}")
                    await asyncio.sleep(current_delay)
                    current_delay *= backoff
        
        return wrapper
    return decorator

@retry(max_attempts=3, delay=1.0)
async def call_external_api(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

Транзакции в БД

Декораторы упрощают управление транзакциями.

from sqlalchemy.ext.asyncio import AsyncSession

def transactional(func):
    """Оборачивает функцию в транзакцию"""
    @functools.wraps(func)
    async def wrapper(session: AsyncSession, *args, **kwargs):
        try:
            result = await func(session, *args, **kwargs)
            await session.commit()
            return result
        except Exception:
            await session.rollback()
            raise
    return wrapper

@transactional
async def transfer_money(session: AsyncSession, from_id: int, to_id: int, amount: decimal.Decimal):
    from_account = await session.get(Account, from_id)
    to_account = await session.get(Account, to_id)
    
    if from_account.balance < amount:
        raise InsufficientFunds()
    
    from_account.balance -= amount
    to_account.balance += amount

Измерение производительности

Для поиска bottleneck'ов использую декораторы профилирования.

import cProfile
import pstats

def profile(func):
    """Профилирует время выполнения функции"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        profiler.enable()
        
        result = func(*args, **kwargs)
        
        profiler.disable()
        stats = pstats.Stats(profiler)
        stats.sort_stats('cumulative')
        stats.print_stats(10)
        
        return result
    return wrapper

@profile
def heavy_computation():
    # Тяжелые вычисления
    pass

Итоги

Декораторы использую для:

• Логирования и мониторинга — отслеживание вызовов
• Кеширования — оптимизация производительности
• Безопасности — проверка прав доступа
• Валидации — проверка входных данных
• Retry логики — обработка временных ошибок
• Управления транзакциями — автоматический commit/rollback
• Профилирования — анализ производительности

Это позволяет писать чистый код, избегая повторений и отделяя перекрёстные заботы от основной логики.