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: str, domain: Any, conn_info: dict[str, Any]) -> None:
    """Initialize Cache with Connection/Adapter details"""
    self.name = name
    self.domain = domain
    self.conn_info = conn_info

    self.ttl = _resolve_ttl(conn_info.get("TTL"), f"Cache '{name}'")

    # Temporary cache of projections
    self._projections: dict[str, type[BaseProjection]] = {}

register_projection

register_projection(
    projection_cls: type[BaseProjection],
) -> None

Registers a projection object for data serialization and de-serialization

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

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

close

close() -> None

Close the cache and release all connections.

Subclasses that hold external resources (connection pools, sockets, etc.) should override this to perform cleanup. The default implementation is a no-op so that adapters without external resources (e.g. the in-memory cache) work without changes.

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

def close(self) -> None:
    """Close the cache and release all connections.

    Subclasses that hold external resources (connection pools, sockets,
    etc.) should override this to perform cleanup.  The default
    implementation is a no-op so that adapters without external
    resources (e.g. the in-memory cache) work without changes.
    """

ping abstractmethod

ping() -> bool

Healthcheck to verify cache is active and accessible

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

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

get_connection abstractmethod

get_connection() -> object

Get the connection object for the cache

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

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

add abstractmethod

add(
    projection: BaseProjection, ttl: TTLValue | None = None
) -> None

Add projection record to cache

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

TTL is in seconds. Accepts a number, or a string holding one, because a TTL sourced from config arrives as a string: environment substitution runs over already-parsed TOML strings. Anything that is not a positive, finite number of seconds raises a ConfigurationError naming the cache.

Omitted (or an empty string) means "use this cache's TTL", which falls back to 300 seconds when the cache configures none.

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

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

@abstractmethod
def add(self, projection: BaseProjection, ttl: TTLValue | None = None) -> None:
    """Add projection record to cache

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

    TTL is in seconds. Accepts a number, or a string holding one, because a
    TTL sourced from config arrives as a string: environment substitution
    runs over already-parsed TOML strings. Anything that is not a positive,
    finite number of seconds raises a `ConfigurationError` naming the cache.

    Omitted (or an empty string) means "use this cache's `TTL`", which falls
    back to 300 seconds when the cache configures none.

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

get abstractmethod

get(key: str) -> BaseProjection | None

Retrieve data by key

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

@abstractmethod
def get(self, key: str) -> BaseProjection | None:
    """Retrieve data by key"""

count abstractmethod

count(key_pattern: str) -> int

Number of entries whose key matches key_pattern.

key_pattern is a glob. See _get_all for the syntax. Like _get_all, this walks the whole keyspace on a store with no native ordering (Redis), so it is O(number of keys) per call.

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

@abstractmethod
def count(self, key_pattern: str) -> int:
    """Number of entries whose key matches `key_pattern`.

    `key_pattern` is a glob. See `_get_all` for the syntax. Like `_get_all`,
    this walks the whole keyspace on a store with no native ordering
    (Redis), so it is O(number of keys) per call.
    """

remove abstractmethod

remove(projection: BaseProjection) -> None

Remove a cache record by projection object

Does nothing if no record exists for the projection.

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

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

    Does nothing if no record exists for the projection.
    """

remove_by_key abstractmethod

remove_by_key(key: str) -> None

Remove a cache record by key

Does nothing if the key is absent.

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

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

    Does nothing if the key is absent.
    """

remove_by_key_pattern abstractmethod

remove_by_key_pattern(key_pattern: str) -> None

Remove cache records by key pattern.

key_pattern is a glob. See _get_all for the syntax.

Does nothing if the pattern matches no keys.

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

@abstractmethod
def remove_by_key_pattern(self, key_pattern: str) -> None:
    """Remove cache records by key pattern.

    `key_pattern` is a glob. See `_get_all` for the syntax.

    Does nothing if the pattern matches no keys.
    """

flush_all abstractmethod

flush_all() -> None

Remove all entries in Cache

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

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

set_ttl abstractmethod

set_ttl(key: str, ttl: TTLValue) -> None

Set a TTL explicitly on a key.

Takes the same shapes as add: a number, or a string holding one, and rejects anything that is not a positive, finite number of seconds, whether or not the key is present.

Otherwise, does nothing if the key is absent.

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

@abstractmethod
def set_ttl(self, key: str, ttl: TTLValue) -> None:
    """Set a TTL explicitly on a key.

    Takes the same shapes as `add`: a number, or a string holding one, and
    rejects anything that is not a positive, finite number of seconds,
    whether or not the key is present.

    Otherwise, does nothing if the key is absent.
    """

get_ttl abstractmethod

get_ttl(key: str) -> float | None

Seconds remaining before key expires.

Seconds, like every other TTL on this port. Stating it is the point: without a unit in the contract, the Redis adapter returned PTTL directly and answered milliseconds while the memory adapter answered seconds, and each adapter's own tests agreed with it (#1307).

Every adapter answers the same three cases:

• None when there is no such key.
• math.inf when the key exists and never expires.
• the seconds remaining otherwise.

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

@abstractmethod
def get_ttl(self, key: str) -> float | None:
    """Seconds remaining before `key` expires.

    Seconds, like every other TTL on this port. Stating it is the point:
    without a unit in the contract, the Redis adapter returned `PTTL`
    directly and answered milliseconds while the memory adapter answered
    seconds, and each adapter's own tests agreed with it (#1307).

    Every adapter answers the same three cases:

    - `None` when there is no such key.
    - `math.inf` when the key exists and never expires.
    - the seconds remaining otherwise.
    """