Кэширование в Python: lru_cache и кастомные декораторы

Зачем кэшировать вычисления в Python?

Вы оптимизировали запрос к базе, но функция вызывает одно и то же дорогое вычисление с одинаковыми аргументами? Кэширование — не только для распределённых систем. Это инструмент, который должен быть в арсенале каждого Python-разработчика. Неправильное использование может создать больше проблем, чем решить.

Основы кэширования

Кэширование сохраняет результат дорогой операции, чтобы последующие вызовы с теми же аргументами возвращали его мгновенно. Плата — память. Хорошая стратегия балансирует скорость и расход памяти, а также корректно инвалидирует кэш. В Python распространённый подход — мемоизация: кэширование возвращаемых значений функций по аргументам.

Простейший кэш — безлимитный, хранит всё навсегда. Это работает для маленьких наборов данных, но приводит к утечкам памяти при больших объёмах. Лучшие стратегии вытесняют старые записи при достижении лимита, используя политики LRU (наименее недавно использованные) или FIFO (первым пришёл — первым ушёл).

Использование functools.lru_cache

Декоратор functools.lru_cache реализует мемоизацию с политикой LRU. Он потокобезопасен, работает с хэшируемыми аргументами и требует минимум настройки. Идеален для чистых функций — тех, что всегда возвращают одинаковый результат для одних и тех же входных данных и не имеют побочных эффектов.

from functools import lru_cache
import time

@lru_cache(maxsize=128)  # Кэшируем до 128 последних вызовов
def expensive_computation(n: int) -> int:
    """Имитация дорогого вычисления."""
    time.sleep(0.1)  # Представим, что это занимает 100 мс
    return n * n

# Первый вызов: 100 мс, вычисляем и кэшируем
result1 = expensive_computation(5)
# Последующие вызовы с тем же аргументом: ~0 мс, возвращаем кэш
result2 = expensive_computation(5)
print(f"Cache info: {expensive_computation.cache_info()}")
# Вывод: CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)

Параметр maxsize контролирует количество хранимых комбинаций аргументов. None создаёт безлимитный кэш — используйте осторожно. Метод cache_info() показывает статистику попаданий/промахов, помогая настроить размер. Для ручной очистки вызовите expensive_computation.cache_clear().

Совет: Используйте lru_cache для рекурсивных алгоритмов, например, чисел Фибоначчи. Без кэша наивная рекурсия имеет экспоненциальную сложность, с кэшем — линейную.

@lru_cache(maxsize=None)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# fibonacci(100) выполняется мгновенно; без кэша зависло бы

Создание кастомных декораторов кэширования

Иногда lru_cache не подходит. Возможно, нужно кэширование с истечением по времени, лимит по памяти или своя логика вытеснения. Создание собственного декоратора даёт полный контроль.

from functools import wraps
import time
from typing import Callable, Any, Dict

def timed_cache(seconds: int = 60):
    """Кэширование результатов на указанное время."""
    def decorator(func: Callable) -> Callable:
        cache: Dict[Any, tuple] = {}  # аргументы -> (результат, временная метка)
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            key = (args, frozenset(kwargs.items()))
            now = time.time()
            if key in cache:
                result, timestamp = cache[key]
                if now - timestamp < seconds:
                    return result  # возвращаем кэш, если не истекло
            result = func(*args, **kwargs)
            cache[key] = (result, now)
            return result
        wrapper.cache_clear = lambda: cache.clear()
        return wrapper
    return decorator

@timed_cache(seconds=30)
def fetch_user_data(user_id: int) -> dict:
    """Имитация API-вызова, кэшируемого на 30 секунд."""
    return {"id": user_id, "name": f"User {user_id}", "timestamp": time.time()}

Этот декоратор истекает через фиксированное время, полезно для данных, меняющихся периодически. Ключ формируется из позиционных и именованных аргументов с использованием frozenset для детерминированного хэширования.

Компромисс: Кастомные декораторы дают гибкость, но требуют внимательной реализации. Потокобезопасность, управление памятью и обработка коллизий ключей — на вас. Начинайте с lru_cache и переходите к кастомным только когда это необходимо.

Кэширование с внешним хранилищем (Redis)

Внутрипроцессное кэширование работает для однопоточных приложений, но распределённым системам нужно общее хранилище. Redis — популярное решение: быстрый, поддерживает истечение нативно, работает на нескольких серверах.

import json
import redis
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

def redis_cache(ttl: int = 300):
    """Кэширование результатов в Redis с TTL в секундах."""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            key = f"{func.__name__}:{args}:{frozenset(kwargs.items())}"
            cached = redis_client.get(key)
            if cached:
                return json.loads(cached)
            result = func(*args, **kwargs)
            redis_client.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@redis_cache(ttl=60)
def external_api_call(endpoint: str) -> dict:
    """Имитация дорогого внешнего API-вызова."""
    return {"endpoint": endpoint, "data": "expensive result"}

Команда setex устанавливает значение с временем истечения, так что Redis автоматически удаляет устаревшие записи. Это безопаснее ручного управления и предотвращает раздувание памяти.

Типичные ошибки

Кэширование изменяемых объектов

Хранение списков, словарей и других изменяемых объектов в кэше может привести к скрытым багам. Если вызывающий код модифицирует возвращённый объект, значение в кэше тоже изменится, влияя на будущие вызовы. Всегда возвращайте копии или используйте неизменяемые структуры.

# НЕПРАВИЛЬНО: возвращаем изменяемый объект из кэша
@lru_cache()
def get_config() -> dict:
    return {"setting": "value"}

config = get_config()
config["setting"] = "modified"  # Повреждаем кэш!

# ПРАВИЛЬНО: возвращаем копию или неизменяемый тип
from copy import deepcopy

@lru_cache()
def get_config_safe() -> dict:
    return deepcopy({"setting": "value"})

Забытая инвалидация кэша

Кэшированные данные со временем устаревают. Если вы кэшируете права пользователя, но не инвалидируете кэш при изменении ролей, пользователи могут сохранить нежелательный доступ. Проектируйте ключи кэша с учётом инвалидации — включайте версию или временную метку, предоставляйте методы для очистки связанных записей.

Избыточное кэширование

Не всё нужно кэшировать. Простые операции могут дольше извлекаться из кэша, чем вычисляться. Профилируйте код перед добавлением кэша и измеряйте эффект. Промах кэша, запускающий исходное вычисление плюс накладные расходы кэша, медленнее, чем отсутствие кэша.

Практический вывод

Эффективное кэширование начинается с понимания, что в вашем коде дорого и безопасно ли повторное использование результатов. Начните с functools.lru_cache для чистых функций, переходите к кэшированию с истечением для данных, меняющихся периодически, и используйте Redis для распределённых систем. Профилируйте до и после — кэширование должно измеримо улучшать производительность, а не только теоретически.

Что делать прямо сейчас:

• Запустите python -m cProfile на медленных функциях, чтобы найти кандидатов на кэширование.
• Изучите документацию functools — декораторы @cache (Python 3.9+) и параметры @lru_cache.
• Если строите распределённые системы, изучите паттерны Redis для вашего сценария.