Кэш Python в памяти со временем жить

Ответ 1

Вы можете использовать модуль expiringdict :

Ядром библиотеки является класс ExpiringDict, представляющий собой упорядоченный словарь с автоматически истекающими значениями для целей кэширования.

В описании они не говорят о многопоточности, поэтому, чтобы не запутаться, используйте Lock.

Ответ 2

OP использует python 2.7, но если вы используете python 3, ExpiringDict упомянутый в принятом ответе, в настоящее время истек. Последний коммит в репозитории github был 17 июня 2017 года, и есть открытая проблема, что он не работает с Python 3.5

Существует более недавно поддерживаемый проект cachetools (последний коммит 14 июня 2018 г.)

pip install cachetools

from cachetools import TTLCache

cache = TTLCache(maxsize=10, ttl=360)
cache['apple'] = 'top dog'
...
>>> cache['apple']
'top dog'
... after 360 seconds...
>>> cache['apple']
KeyError exception thrown

ttl время жить в секундах.

Ответ 3

Если вы не хотите использовать какие-либо третьи библиотеки, вы можете добавить еще один параметр в вашу дорогую функцию: ttl_hash=None. Этот новый параметр называется так называемым "чувствительным ко времени хэшем", его единственная цель - повлиять на lru_cache.

Например:

from functools import lru_cache
import time


@lru_cache()
def my_expensive_function(a, b, ttl_hash=None):
    del ttl_hash  # to emphasize we don't use it and to shut pylint up
    return a + b  # horrible CPU load...

def get_ttl_hash(seconds=3600):
    """Return the same value withing 'seconds' time period"""
    return round(time.time() / seconds)

# somewhere in your code...
res = my_expensive_function(2, 2, ttl_hash=get_ttl_hash())
# cache will be updated once in an hour

Ответ 4

Что касается кэш-памяти с истекающим сроком действия в памяти, то для общего назначения обычно используется шаблон проектирования, который обычно делается не через словарь, а через декоратор функции или метода. Кеш-словарь управляется за кулисами. Таким образом, этот ответ несколько дополняет ответ пользователя, который использует словарь, а не декоратор.

ttl_cache декоратор в cachetools==3.1.0 работает во многом как functools.lru_cache, но с временем жизни.

import cachetools.func

@cachetools.func.ttl_cache(maxsize=128, ttl=10 * 60)
def example_function(key):
    return get_expensively_computed_value(key)


class ExampleClass:
    EXP = 2

    @classmethod
    @cachetools.func.ttl_cache()
    def example_classmethod(cls, i):
        return i * cls.EXP

    @staticmethod
    @cachetools.func.ttl_cache()
    def example_staticmethod(i):
        return i * 3

Ответ 5

Что-то вроде этого?

from time import time, sleep
import itertools
from threading import Thread, RLock
import signal


class CacheEntry():
  def __init__(self, string, ttl=20):
    self.string = string
    self.expires_at = time() + ttl
    self._expired = False

  def expired(self):
    if self._expired is False:
      return (self.expires_at < time())
    else:
      return self._expired

class CacheList():
  def __init__(self):
    self.entries = []
    self.lock = RLock()

  def add_entry(self, string, ttl=20):
    with self.lock:
        self.entries.append(CacheEntry(string, ttl))

  def read_entries(self):
    with self.lock:
        self.entries = list(itertools.dropwhile(lambda x:x.expired(), self.entries))
        return self.entries

def read_entries(name, slp, cachelist):
  while True:
    print "{}: {}".format(name, ",".join(map(lambda x:x.string, cachelist.read_entries())))
    sleep(slp)

def add_entries(name, ttl, cachelist):
  s = 'A'
  while True:
    cachelist.add_entry(s, ttl)
    print("Added ({}): {}".format(name, s))
    sleep(1)
    s += 'A'



if __name__ == "__main__":
  signal.signal(signal.SIGINT, signal.SIG_DFL)

  cl = CacheList()
  print_threads = []
  print_threads.append(Thread(None, read_entries, args=('t1', 1, cl)))
  # print_threads.append(Thread(None, read_entries, args=('t2', 2, cl)))
  # print_threads.append(Thread(None, read_entries, args=('t3', 3, cl)))

  adder_thread = Thread(None, add_entries, args=('a1', 2, cl))
  adder_thread.start()

  for t in print_threads:
    t.start()

  for t in print_threads:
    t.join()

  adder_thread.join()

Ответ 6

Я очень люблю идею от @iutinvg, я просто хотел бы пойти немного дальше. Отсоедините его от необходимости знать, как пройти ttl, и просто сделайте его декоратором, чтобы вам не приходилось об этом думать. Если у вас есть django, py3 и вы не чувствуете, что pip устанавливает какие-либо зависимости, попробуйте это.

import time
from django.utils.functional import lazy
from functools import lru_cache, partial, update_wrapper


def lru_cache_time(seconds, maxsize=None):
    """
    Adds time aware caching to lru_cache
    """
    def wrapper(func):
        # Lazy function that makes sure the lru_cache() invalidate after X secs
        ttl_hash = lazy(lambda: round(time.time() / seconds), int)()

        @lru_cache(maxsize)
        def time_aware(__ttl, *args, **kwargs):
            """
            Main wrapper, note that the first argument ttl is not passed down. 
            This is because no function should bother to know this that 
            this is here.
            """
            def wrapping(*args, **kwargs):
                return func(*args, **kwargs)
            return wrapping(*args, **kwargs)
        return update_wrapper(partial(time_aware, ttl_hash), func)
    return wrapper


@lru_cache_time(seconds=10)
def meaning_of_life():
    """
    This message should show up if you call help().
    """
    print('this better only show up once!')
    return 42


@lru_cache_time(seconds=10)
def mutiply(a, b):
    """
    This message should show up if you call help().
    """
    print('this better only show up once!')
    return a * b

# This is a test, prints a '.' for every second, there should be 10s 
# beween each "this better only show up once!" *2 because of the two functions.
for _ in range(20):
    meaning_of_life()
    mutiply(50, 99991)
    print('.')
    time.sleep(1)