Skip to content

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 
 9
10
11
12
13
14
15
16
17
18
19
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 
21
22
23
24
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

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 
26
27
28
29
30
31
32
33
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 
35
36
37
@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 
39
40
41
@abstractmethod
def get_connection(self) -> object:
    """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 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@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: str) -> Optional[BaseProjection]

Retrieve data by key

Source code in src/protean/port/cache.py 
61
62
63
@abstractmethod
def get(self, key: str) -> Optional[BaseProjection]:
    """Retrieve data by key"""

get_all abstractmethod 

get_all(
    key_pattern: str, last_position: int = 0, size: int = 25
) -> list[BaseProjection]

Retrieve data by key pattern

Source code in src/protean/port/cache.py 
65
66
67
68
69
@abstractmethod
def get_all(
    self, key_pattern: str, last_position: int = 0, size: int = 25
) -> list[BaseProjection]:
    """Retrieve data by key pattern"""

count abstractmethod 

count(key_pattern: str) -> int

Retrieve count of data by key pattern

Source code in src/protean/port/cache.py 
71
72
73
@abstractmethod
def count(self, key_pattern: str) -> int:
    """Retrieve count of data by key pattern"""

remove abstractmethod 

remove(projection: BaseProjection) -> None

Remove a cache record by projection object

Source code in src/protean/port/cache.py 
75
76
77
@abstractmethod
def remove(self, projection: BaseProjection) -> None:
    """Remove a cache record by projection object"""

remove_by_key abstractmethod 

remove_by_key(key: str) -> None

Remove a cache record by key

Source code in src/protean/port/cache.py 
79
80
81
@abstractmethod
def remove_by_key(self, key: str) -> None:
    """Remove a cache record by key"""

remove_by_key_pattern abstractmethod 

remove_by_key_pattern(key_pattern: str) -> None

Remove a cache record by key pattern

Source code in src/protean/port/cache.py 
83
84
85
@abstractmethod
def remove_by_key_pattern(self, key_pattern: str) -> None:
    """Remove a cache record by key pattern"""

flush_all abstractmethod 

flush_all() -> None

Remove all entries in Cache

Source code in src/protean/port/cache.py 
87
88
89
@abstractmethod
def flush_all(self) -> None:
    """Remove all entries in Cache"""

set_ttl abstractmethod 

set_ttl(key: str, ttl: Union[int, float]) -> None

Set a TTL explicitly on a key

Source code in src/protean/port/cache.py 
91
92
93
@abstractmethod
def set_ttl(self, key: str, ttl: Union[int, float]) -> None:
    """Set a TTL explicitly on a key"""

get_ttl abstractmethod 

get_ttl(key: str) -> float

Get the TTL set on a key

Source code in src/protean/port/cache.py 
95
96
97
@abstractmethod
def get_ttl(self, key: str) -> float:
    """Get the TTL set on a key"""