BaseCache

Cache interface for read-optimized storage. All cache adapters (Memory, Redis, etc.) implement this contract.

See Cache Adapters for concrete adapter configuration.

Initialize Cache with Connection/Adapter details

Source code in src/protean/port/cache.py

def __init__(self, name, domain, conn_info: dict):
    """Initialize Cache with Connection/Adapter details"""
    self.name = name
    self.domain = domain
    self.conn_info = conn_info

    # Default TTL: 300 seconds
    self.ttl = conn_info.get("TTL", 300)

    # Temporary cache of projections
    self._projections = {}

register_projection

register_projection(projection_cls)

Registers a projection object for data serialization and de-serialization

Source code in src/protean/port/cache.py

def register_projection(self, projection_cls):
    """Registers a projection object for data serialization and de-serialization"""
    projection_name = underscore(projection_cls.__name__)
    self._projections[projection_name] = projection_cls

ping abstractmethod

ping()

Healthcheck to verify cache is active and accessible

Source code in src/protean/port/cache.py

@abstractmethod
def ping(self):
    """Healthcheck to verify cache is active and accessible"""

get_connection abstractmethod

get_connection()

Get the connection object for the cache

Source code in src/protean/port/cache.py

@abstractmethod
def get_connection(self):
    """Get the connection object for the cache"""

add abstractmethod

add(projection: BaseProjection, ttl: Optional[Union[int, float]] = None) -> None

Add projection record to cache

KEY: Projection ID Value: Projection Data (derived from to_dict())

TTL is in seconds. If not specified explicitly in method call, it can be picked up from broker configuration. In the absence of configuration, it can be defaulted to an optimum value, say 300 seconds.

PARAMETER	DESCRIPTION
projection	Projection Instance containing data TYPE: BaseProjection
ttl	Timeout in seconds. Defaults to None. TYPE: (int, float) DEFAULT: None

Source code in src/protean/port/cache.py

@abstractmethod
def add(
    self, projection: BaseProjection, ttl: Optional[Union[int, float]] = None
) -> None:
    """Add projection record to cache

    KEY: Projection ID
    Value: Projection Data (derived from `to_dict()`)

    TTL is in seconds. If not specified explicitly in method call,
    it can be picked up from broker configuration. In the absence of
    configuration, it can be defaulted to an optimum value, say 300 seconds.

    Args:
        projection (BaseProjection): Projection Instance containing data
        ttl (int, float, optional): Timeout in seconds. Defaults to None.
    """

get abstractmethod

get(key)

Retrieve data by key

Source code in src/protean/port/cache.py

@abstractmethod
def get(self, key):
    """Retrieve data by key"""

get_all abstractmethod

get_all(key_pattern, last_position=0, size=25)

Retrieve data by key pattern

Source code in src/protean/port/cache.py

@abstractmethod
def get_all(self, key_pattern, last_position=0, size=25):
    """Retrieve data by key pattern"""

count abstractmethod

count(key_pattern)

Retrieve count of data by key pattern

Source code in src/protean/port/cache.py

@abstractmethod
def count(self, key_pattern):
    """Retrieve count of data by key pattern"""

remove abstractmethod

remove(projection)

Remove a cache record by projection object

Source code in src/protean/port/cache.py

@abstractmethod
def remove(self, projection):
    """Remove a cache record by projection object"""

remove_by_key abstractmethod

remove_by_key(key)

Remove a cache record by key

Source code in src/protean/port/cache.py

@abstractmethod
def remove_by_key(self, key):
    """Remove a cache record by key"""

remove_by_key_pattern abstractmethod

remove_by_key_pattern(key_pattern)

Remove a cache record by key pattern

Source code in src/protean/port/cache.py

@abstractmethod
def remove_by_key_pattern(self, key_pattern):
    """Remove a cache record by key pattern"""

flush_all abstractmethod

flush_all()

Remove all entries in Cache

Source code in src/protean/port/cache.py

@abstractmethod
def flush_all(self):
    """Remove all entries in Cache"""

set_ttl abstractmethod

set_ttl(key, ttl)

Set a TTL explicitly on a key

Source code in src/protean/port/cache.py

@abstractmethod
def set_ttl(self, key, ttl):
    """Set a TTL explicitly on a key"""

get_ttl abstractmethod

get_ttl(key)

Get the TTL set on a key

Source code in src/protean/port/cache.py

@abstractmethod
def get_ttl(self, key):
    """Get the TTL set on a key"""