Skip to content

IM-Cloud-Spain-Connectors/python-cache-interface

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

15 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Python Cache Interface

Test

Simple and clean Cache interface for python projects.

Installation

The easiest way to install the Cache Interface is to get the latest version from PyPI:

# using poetry
poetry add rndi-cache-interface
# using pip
pip install rndi-cache-interface

The Contract

This package provides one simple contract or interface: Cache.

class Cache(metaclass=ABCMeta):  # pragma: no cover
    @abstractmethod
    def has(self, key: str) -> bool:
        """
        Check if the given key exists in the cache.
        :param key: str The key to search for.
        :return: bool True if the key exists, false otherwise (expired
                      values returns False).
        """

    @abstractmethod
    def get(self, key: str, default: Any = None) -> Any:
        """
        Get the required value by key from the cache.
        :param key: str The key of the stored value.
        :param default: Any The default value to use if the key has no value in
                        cache, None by default. If the default value is a
                        callable, the result value of the execution will
                        be stored in the cache. The expiration time is 900
                        seconds. The callable may return a Tuple[Any, int]
                        with the actual value as first element and the desired
                        expired time as second value.
        :return: Any The requested value by key.
        """

    @abstractmethod
    def put(self, key: str, value: Any, ttl: Optional[int] = None) -> Any:
        """
        Put a value by key in the cache.
        :param key: str The key to store the value.
        :param value: Any The value to store in cache
        :param ttl: int Expiration time in seconds
        :return: Any The stored value by key.
        """

    @abstractmethod
    def delete(self, key: str) -> None:
        """
        Delete a value by key from the cache.
        :param key: str The key to be deleted.
        :return:
        """

    @abstractmethod
    def flush(self, expired_only: bool = False) -> None:
        """
        Delete all the items in the cache.
        :param expired_only: bool If True only the expired values will be deleted.
        :return:
        """

The Adapter

Just initialize the class you want and use the public methods:

from rndi.cache.contracts import Cache
from rndi.cache.adapters.fs.adapter import FileSystemCacheAdapter


def some_process_that_requires_cache(cache: Cache):
    # retrieve the data from cache, ir the key is not cached yet and the default 
    # value is a callable the cache will use it to compute and cache the value
    user = cache.get('user-id', lambda: db_get_user('user-id'))

    print(user)


# inject the desired cache adapter.
cache = FileSystemCacheAdapter('/tmp', 900)
some_process_that_requires_cache(cache)

Checking the if the key exists in the cache.

cache.has('user-id')

Getting a value from the cache.

# will return None if the value not exists in the cache.
cache.get('user-id')

# if the key is not present, the default value will be used.
cache.get('user-id', {'id': 'user-id', 'email': 'vincent.vega@mail.com'})

# if the default value is a callable the cache will use the callable to 
# compute and cache the request value.
cache.get('user-id', lambda: db_get_user('user-id'))

# you can also provide custom ttl (time to live) for a computed value by 
# returning a tuple of computed value and the ttl integer (Tuple[Any, int]).
cache.get('user-id', lambda: (db_get_user('user-id'), 3600))

Storing value by key in cache.

# store the given value by key.
cache.put('user-id', {'id': 'user-id', 'email': 'vincent.vega@mail.com'})

# store the given value with custom ttl (time to live).
cache.put('user-id', {'id': 'user-id', 'email': 'vincent.vega@mail.com'}, 3600)

Delete a value from cache by key.

cache.delete('user-id')

Flush the complete cache.

cache.flush()

Available Adapters

This package provides along with the contract, two small adapters:

  • NullCacheAdapter that will never cache anything.
  • FileSystemCacheAdapter very simple file-based cache.