Python functools lru_cache

22 Дек 2024 , 77

Для кеширование результатов функции в Python есть декоратор lru_cache из стандартной библиотеки functools.

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

Кеширование

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

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

  • Функция является детерминированной и для одинаковых входных данных всегда возвращает одно и то же значение.

  • Функция возвращает недетерминированное значение, но оно остается по­лезным и действительным в течение некоторого времени.

Иначе говоря, детерминированная функция всегда возвращает один и тот же результат для одного набора аргументов, а недетерминированная функция воз­вращает результаты, которые могут меняться со временем.

Детермированные функции

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

Рассмотрим функцию вычисления n-числа Фибоначчи . Эта функция детермированная , потому что сколько бы мы ее не вызывали она будет возвращать одинаковый результат для конкретных входных параметров.


def fibonacci(n):
    print(f"---Вызов функции вычисления {n=} элемента Фибоначчи----")
    if n < 0:
        raise ValueError("n должно быть неотрицательным целым числом")
    elif n == 0:
        return 0
    elif n == 1:
        return 1

    a, b = 0, 1  # Инициализируем первые два числа Фибоначчи
    for _ in range(2, n + 1):
        a, b = b, a + b  # Обновляем значения

    return b


print(f"10 число в последовательности Фибоначчи равно = {fibonacci(n=10)}")
print(f"10 число в последовательности Фибоначчи равно = {fibonacci(n=10)}")
print(f"15 число в последовательности Фибоначчи равно = {fibonacci(n=15)}")
print(f"10 число в последовательности Фибоначчи равно = {fibonacci(n=10)}")
print(f"15 число в последовательности Фибоначчи равно = {fibonacci(n=15)}")

Результат запуска кода выше:


---Вызов функции вычисления n=10 элемента Фибоначчи----
10 число в последовательности Фибоначчи равно = 55
---Вызов функции вычисления n=10 элемента Фибоначчи----
10 число в последовательности Фибоначчи равно = 55
---Вызов функции вычисления n=15 элемента Фибоначчи----
15 число в последовательности Фибоначчи равно = 610
---Вызов функции вычисления n=10 элемента Фибоначчи----
10 число в последовательности Фибоначчи равно = 55
---Вызов функции вычисления n=15 элемента Фибоначчи----
15 число в последовательности Фибоначчи равно = 610

Как видно выше для конкретного входного параметра n возвращается всегда один и тот же результат и поэтому мы можем применить кеширование. И для этого будем использовать декоратор lru_cache

Добавим декоратор lru_cache и посмотрим на вывод.


from functools import lru_cache


@lru_cache
def fibonacci(n):
    print(f"---Вызов функции вычисления {n=} элемента Фибоначчи----")
    if n < 0:
        raise ValueError("n должно быть неотрицательным целым числом")
    elif n == 0:
        return 0
    elif n == 1:
        return 1

    a, b = 0, 1  # Инициализируем первые два числа Фибоначчи
    for _ in range(2, n + 1):
        a, b = b, a + b  # Обновляем значения

    return b


print(f"10 число в последовательности Фибоначчи равно = {fibonacci(n=10)}")
print(f"10 число в последовательности Фибоначчи равно = {fibonacci(n=10)}")
print(f"15 число в последовательности Фибоначчи равно = {fibonacci(n=15)}")
print(f"10 число в последовательности Фибоначчи равно = {fibonacci(n=10)}")
print(f"15 число в последовательности Фибоначчи равно = {fibonacci(n=15)}")


Запустим код выше с декоратормlru_cache


---Вызов функции вычисления n=10 элемента Фибоначчи----
10 число в последовательности Фибоначчи равно = 55
10 число в последовательности Фибоначчи равно = 55
---Вызов функции вычисления n=15 элемента Фибоначчи----
15 число в последовательности Фибоначчи равно = 610
10 число в последовательности Фибоначчи равно = 55
15 число в последовательности Фибоначчи равно = 610

Как видим выше , при входном n равном 10 первый раз вызывается функция вычисления числа Фибоначчи , но при втором вызове n=10 уже эта функция не вызывается и результат берется из кеша , далее для n=15 мы опять вызываем функцию для вычисления числа Фибоначчи , так как для 15 число мы не вычисляли и его нет в кеше , а далее при повторных вызовах мы уже берем результат из кеша и функция не вызывается

Декоратор lru_cache

Дeкopaтop lru_cache создает кэш возвращаемых значений заданной функции, работающий по принципу LRU (Least Recently Used, «вытеснение давно не использовавшихся элементов•). Он перехватывает входные аргументы функции и сравнивает их со списком недавно использованных наборов аргументов. Если обнаружится совпадение, то вместо вызова декорируемой функции возвраща­ется кэшированное значение. Если совпадений нет, то сначала вызывается ис­ходная функция, а затем возвращенное значение сохраняется в кэше для ис­пользования в будущем.

Дeкopaтop lru_cache автоматически ведёт в памяти словарь (кеш), где:

  • Ключ: кортеж входных аргументов функции (с учётом их порядка и именованных параметров).

  • Значение: уже вычисленный результат функции.

Дeкopaтop lru_cache реализует механизм memoization (мемоизации) для детерминированных функций.

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

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

При вызове lru_cache() вы можете указать несколько параметров:

  1. maxsize (по умолчанию 128)

    • Максимальное количество уникальных результатов, которые могут храниться в кэше.

    • Если лимит превышен, используется политика «Least Recently Used» (LRU), то есть из кэша будут удаляться самые «давно используемые» записи, чтобы освободить место для новых.

    • Если указать None, кэш будет иметь неограниченный размер, что может привести к высоким затратам по памяти.

  2. typed (по умолчанию False)

    • Если typed=True, то аргументы, различающиеся только типом, будут разными ключами для кэша.Например, func(3) и func(3.0) будут считаться разными вызовами и кэшироваться отдельно.

    • Если typed=False, вызов функции с аргументами 3 и 3.0 будет считаться одним и тем же и вернёт одно и то же значение из кэша.

    Недетерминированные функции

    Недетерминированные функции — функции, у которых результат может меняться даже при одинаковых входных данных.

    Причины могут быть разные:

    • Зависимость от текущего времени.

    • Зависимость от внешнего состояния (например, содержимое базы данных, состояние сети).

    • Внешние API-запросы, которые могут возвращать разные результаты.

    Соответственно, если мы кешируем такую функцию, мы можем столкнуться с ситуацией, когда кеш возвращает уже неактуальный результат.

    Некоторые примеры недетерминированной функции

    • Функция, возвращающая курс валют (приходит с внешнего сервера и может меняться со временем).

    • Функция, достающая «список последних сообщений» из базы данных.

    Можем ли мы кешировать результаты таких функций с помощью lru_cache? Да мы можем кешировать при определенных данных , но нужно учитывать риск получения устаревших (некорректных данных) и сложности инвалидации кеша.

    Рассмотрим пример недетерминированной функции, которая получает внешний (меняющийся) курс Bitcoin к доллару США с сайта CoinDesk

    
    import requests
    
    
    def get_bitcoin_price_usd() -> float:
        response = requests.get("https://api.coindesk.com/v1/bpi/currentprice/BTC.json")
        response.raise_for_status()
        data = response.json()
        return data["bpi"]["USD"]["rate_float"]
    
    

    И мы хотим закешировать результат на 1 час (3600 секунд). То есть если функция будет вызываться многократно в течение одного часа, она вернёт один и тот же (закэшированный) результат, а после истечения часа — автоматически обновит данные. Но к сожалению , lru_cache по умолчанию не имеет механизма «протухания» - TTL(Time To Live). То есть, если вы не сбросите кэш вручную, данные будут «вечны» в рамках работы приложения.

    Но мы же программисты и мы можем реализовать декоратор, который добавляет «протухание» (TTL) к стандартному lru_cache

    Полный листинг

    
    import requests
    import functools
    import time
    
    
    def ttl_lru_cache(ttl: float):
        """
        Декоратор, который добавляет «протухание» (TTL) к стандартному lru_cache.
        Как только проходит 'ttl' секунд с момента первого кеширования,
        мы сбрасываем кэш при следующем вызове.
        """
        def decorator(func):
    
            cached_func = functools.lru_cache(maxsize=None)(func)
            cached_func._time_cached = None
            
            @functools.wraps(cached_func)
            def wrapper(*args, **kwargs):
                # Если ещё нет времени кеширования (кэш пуст) — просто вызываем
                if cached_func._time_cached is None:
                    cached_func._time_cached = time.time()
    
                # Если прошло больше, чем ttl секунд — сбрасываем кэш и обновляем время
                if (time.time() - cached_func._time_cached) > ttl:
                    cached_func.cache_clear()
                    cached_func._time_cached = time.time()
    
                return cached_func(*args, **kwargs)
            
            wrapper.cache_clear = cached_func.cache_clear
            wrapper.cache_info = cached_func.cache_info
            return wrapper
    
        return decorator
    
    
    # кешируем на 1 час 
    @ttl_lru_cache(ttl=60*60)
    def get_bitcoin_price_usd() -> float:
        print("Вызов функции получения курса Биткоина")
        response = requests.get("https://api.coindesk.com/v1/bpi/currentprice/BTC.json")
        response.raise_for_status()
        data = response.json()
        return data["bpi"]["USD"]["rate_float"]
    
    
    print(get_bitcoin_price_usd())
    print(get_bitcoin_price_usd())
    print(get_bitcoin_price_usd())
    
    

    Но вместо создания ручного декоратора с поддержкой TTL для декоратора lru_cache я бы вам посоветовал использовать библиотеку cachetools.

    
    pip install cachetools
    
    

    То же самое поведение , что и выше мы можем получить следующим способом

    
    from cachetools import cached, TTLCache
    
    
    @cached(TTLCache(maxsize=128, ttl=3600))
    def get_bitcoin_price_usd() -> float:
        print("Вызов функции получения курса Биткоина")
        response = requests.get("https://api.coindesk.com/v1/bpi/currentprice/BTC.json")
        response.raise_for_status()
        data = response.json()
        return data["bpi"]["USD"]["rate_float"]
    
    
    print(get_bitcoin_price_usd())
    print(get_bitcoin_price_usd())
    print(get_bitcoin_price_usd())
    
    

    Библиотека cachetools предоставляет мощные и гибкие инструменты для реализации кэширования в Python, включая поддержку TTL. :

    Использование cachetools особенно рекомендуется в случаях, когда требуется:

    • Кэширование с ограничением по времени: Данные актуальны только на определённый промежуток времени.

    • Управление размером кэша: Контроль над количеством хранимых записей.

    • Различные стратегии кэширования: Например, LRU, LFU и др

    Нужно отметить , что lru_cache может работать только с синхронными функциями и методами. Если ваша функция объявлена как async def, прямое использование @lru_cache не подойдёт, т.к. при вызове асинхронных функций нужен другой механизм выполнения (через await), а lru_cache ожидает обычный» вызов. Для кеширования(мемоизации) асинхронных функций и методов я посоветовал бы библиотеку async-lru. Это практически lru_cache для асинхронных функций , но с поддержкой TTL

    Также хотел бы отметить такой момент , что когда вы декорируете метод класса при помощи @lru_cache, следует помнить, что self входит в список аргументов, формирующих ключ кэша: Если цель была «один общий кэш для всех экземпляров класса», то так не получится, потому что self участвует в хешировании, и для каждого объекта будет «свой» набор сохранённых результатов. Поэтому если у вас есть методы класса и вы хотите просто навесить этот декоратор , то учитываейте этот момент

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

    Заключение

    В этой статье мы рассмотрели декоратор lru_cache из стандартной библиотеки. Поняли , что его уместно использовать для мемоизации данных для детермированных функций. Для недетермированных функций использование данного декоратора не совсем целесообразно , так как нам приходится вручную управлять «протуханием» кэша , так как нет TTL. Да мы привели пример как добавить вручную TTL для декоратора lru_cache , но для этого лучше использовать сторонние решения, такие как cachetools

    Декоратор lru_cache работает только с синхронными функциями и методами. Если ваша функция объявлена как async def, прямое использование @lru_cache не подойдёт, т.к. при вызове асинхронных функций нужен другой механизм выполнения (через await), а lru_cache ожидает «обычный» вызов.

    Когда вы декорируете метод класса при помощи @lru_cache, следует помнить, что self входит в список аргументов, формирующих ключ кэша:Если цель была «один общий кэш для всех экземпляров класса», то так не получится, потому что self участвует в хешировании, и для каждого объекта будет «свой» набор сохранённых результатов. Поэтому если у вас есть методы класса и вы хотите просто навесить этот декоратор , то учитываейте этот момент.

    Буду рад любой критике и к любым замечаниям

    comments powered by Disqus

Подписка

Подпишитесь на наш список рассылки, чтобы получать обновления из блога

Рубрики

Теги