Skip to content

Cache Interfaces

Abstract interfaces defining sync and async cache operations.

Synchronous Interface

CacheInterface

CacheInterface

Bases: ABC

Abstract cache interface supporting sync operations.

Source code in src/jinpy_utils/cache/interfaces.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class CacheInterface(ABC):
    """Abstract cache interface supporting sync operations."""

    @abstractmethod
    def get(self, key: str) -> Any | None: ...

    @abstractmethod
    def set(self, key: str, value: Any, ttl: float | None = None) -> None: ...

    @abstractmethod
    def delete(self, key: str) -> None: ...

    @abstractmethod
    def exists(self, key: str) -> bool: ...

    @abstractmethod
    def clear(self) -> None: ...

    @abstractmethod
    def get_many(self, keys: list[str]) -> dict[str, Any | None]: ...

    @abstractmethod
    def set_many(self, items: Mapping[str, Any], ttl: float | None = None) -> None: ...

    @abstractmethod
    def delete_many(self, keys: list[str]) -> None: ...

    @abstractmethod
    def incr(self, key: str, amount: int = 1, ttl: float | None = None) -> int: ...

    @abstractmethod
    def decr(self, key: str, amount: int = 1, ttl: float | None = None) -> int: ...

    @abstractmethod
    def ttl(self, key: str) -> float | None: ...

    @abstractmethod
    def touch(self, key: str, ttl: float) -> None: ...

    @abstractmethod
    def is_healthy(self) -> bool: ...

    @abstractmethod
    def close(self) -> None: ...

Asynchronous Interface

AsyncCacheInterface

AsyncCacheInterface

Bases: ABC

Abstract cache interface supporting async operations.

Source code in src/jinpy_utils/cache/interfaces.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class AsyncCacheInterface(ABC):
    """Abstract cache interface supporting async operations."""

    @abstractmethod
    async def aget(self, key: str) -> Any | None: ...

    @abstractmethod
    async def aset(self, key: str, value: Any, ttl: float | None = None) -> None: ...

    @abstractmethod
    async def adelete(self, key: str) -> None: ...

    @abstractmethod
    async def aexists(self, key: str) -> bool: ...

    @abstractmethod
    async def aclear(self) -> None: ...

    @abstractmethod
    async def aget_many(self, keys: list[str]) -> dict[str, Any | None]: ...

    @abstractmethod
    async def aset_many(
        self, items: Mapping[str, Any], ttl: float | None = None
    ) -> None: ...

    @abstractmethod
    async def adelete_many(self, keys: list[str]) -> None: ...

    @abstractmethod
    async def aincr(
        self, key: str, amount: int = 1, ttl: float | None = None
    ) -> int: ...

    @abstractmethod
    async def adecr(
        self, key: str, amount: int = 1, ttl: float | None = None
    ) -> int: ...

    @abstractmethod
    async def attl(self, key: str) -> float | None: ...

    @abstractmethod
    async def atouch(self, key: str, ttl: float) -> None: ...

    @abstractmethod
    async def ais_healthy(self) -> bool: ...

    @abstractmethod
    async def aclose(self) -> None: ...

Notes

These interfaces define the contract implemented by concrete backends such as in-memory, file, and Redis backends.