Skip to content

Cache Core API

Core cache classes and helper functions providing synchronous and asynchronous APIs, a singleton-aware manager, and context helpers.

Factory Functions

get_cache

get_cache 

get_cache(backend: str | None = None) -> Cache

Return a Cache facade bound to the given backend (or default).

Source code in src/jinpy_utils/cache/core.py 
660
661
662
663
664
def get_cache(backend: str | None = None) -> Cache:
    """
    Return a Cache facade bound to the given backend (or default).
    """
    return Cache(backend=backend)

get_cache_manager

get_cache_manager 

get_cache_manager(config: CacheManagerConfig | None = None) -> CacheManager

Return the singleton CacheManager, optionally reconfiguring with a new config. Reconfiguration is applied if a non-None config is passed.

Source code in src/jinpy_utils/cache/core.py 
667
668
669
670
671
672
673
674
675
676
677
678
679
def get_cache_manager(
    config: CacheManagerConfig | None = None,
) -> CacheManager:
    """
    Return the singleton CacheManager,
    optionally reconfiguring with a new config.
    Reconfiguration is applied if a non-None config is passed.
    """
    mgr = CacheManager()
    if config is not None:
        # Reconfigure existing instance
        mgr._reconfigure(config)
    return mgr

Core Classes

CacheManager

CacheManager

Singleton-aware cache manager with async/sync APIs.

Responsibilities: - Initialize and manage multiple cache backends (memory, file, redis, etc.) - Provide a consistent sync and async API delegating to selected backend - Enforce configuration via Pydantic models - Follow SOLID principles (separation of concerns, single responsibility) - 12-Factor: environment-driven config handled by the Pydantic config layer

Usage

manager = CacheManager() # uses defaults (in-memory) manager.set("key", {"a": 1}, ttl=60) value = manager.get("key")

Async

await manager.aset("k", "v", ttl=30) v = await manager.aget("k")

Source code in src/jinpy_utils/cache/core.py 
 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
 54
 55
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
class CacheManager:
    """
    Singleton-aware cache manager with async/sync APIs.

    Responsibilities:
    - Initialize and manage multiple cache backends (memory, file, redis, etc.)
    - Provide a consistent sync and async API delegating to selected backend
    - Enforce configuration via Pydantic models
    - Follow SOLID principles (separation of concerns, single responsibility)
    - 12-Factor: environment-driven config handled by the Pydantic config layer

    Usage:
        manager = CacheManager()  # uses defaults (in-memory)
        manager.set("key", {"a": 1}, ttl=60)
        value = manager.get("key")

        # Async
        await manager.aset("k", "v", ttl=30)
        v = await manager.aget("k")
    """

    _instance: CacheManager | None = None
    _lock = threading.Lock()

    def __new__(
        cls,
        config: CacheManagerConfig | None = None,
    ) -> CacheManager:
        # Strict singleton: if instance exists, return it;
        # first call constructs.
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, config: CacheManagerConfig | None = None) -> None:
        # Idempotent init to support singleton semantics.
        if getattr(self, "_initialized", False):
            # If new config provided later, allow reconfigure explicitly
            if config is not None:
                self._reconfigure(config)
            return

        self._initialized = True
        self.config = config or CacheManagerConfig()
        self._backends: dict[str, BaseBackend] = {}
        self._default_backend_name: str | None = self.config.default_backend
        self._initialize_backends()

    def _reconfigure(self, config: CacheManagerConfig) -> None:
        """
        Reconfigure the manager with a new configuration.

        Safely closes previous backends and re-initializes with the new config.
        """
        self.close()  # best-effort close of existing
        self.config = config
        self._backends.clear()
        self._default_backend_name = self.config.default_backend
        self._initialize_backends()

    def _initialize_backends(self) -> None:
        """
        Initialize and register all enabled backends per configuration.

        Ensures a default backend is available.
        """
        # Provide a default in-memory backend if none configured.
        if not self.config.backends:
            default = MemoryCacheConfig(name="default_memory")
            self.config.backends = [default]
            self._default_backend_name = "default_memory"

        for backend_cfg in self.config.backends:
            if getattr(backend_cfg, "enabled", True):
                backend = CacheBackendFactory.create(backend_cfg)
                self._backends[backend_cfg.name] = backend

        if not self._backends:
            raise CacheConfigurationError(
                "No enabled cache backends configured",
                config_section="backends",
            )

        if not self._default_backend_name:
            # pick first enabled backend as default
            self._default_backend_name = next(iter(self._backends.keys()))

        if self._default_backend_name not in self._backends:
            raise CacheConfigurationError(
                "Default backend name not found",
                config_section="default_backend",
                config_value=str(self._default_backend_name),
            )

    def get_backend(self, name: str | None = None) -> BaseBackend:
        """
        Retrieve a backend by name, or the default if name is None.
        """
        bname = name or self._default_backend_name
        if not bname or bname not in self._backends:
            raise CacheConfigurationError(
                "Requested backend not found",
                config_section="backend",
                config_value=str(bname),
            )
        return self._backends[bname]

    # --------------------------- Sync API --------------------------- #

    def get(self, key: str, *, backend: str | None = None) -> Any | None:
        return self.get_backend(backend).get(key)

    def set(
        self,
        key: str,
        value: Any,
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> None:
        self.get_backend(backend).set(key, value, ttl)

    def delete(self, key: str, *, backend: str | None = None) -> None:
        self.get_backend(backend).delete(key)

    def exists(self, key: str, *, backend: str | None = None) -> bool:
        return self.get_backend(backend).exists(key)

    def clear(self, *, backend: str | None = None) -> None:
        self.get_backend(backend).clear()

    def get_many(
        self, keys: list[str], *, backend: str | None = None
    ) -> dict[str, Any | None]:
        return self.get_backend(backend).get_many(keys)

    def set_many(
        self,
        items: dict[str, Any],
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> None:
        self.get_backend(backend).set_many(items, ttl)

    def delete_many(
        self,
        keys: list[str],
        *,
        backend: str | None = None,
    ) -> None:
        self.get_backend(backend).delete_many(keys)

    def incr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> int:
        return self.get_backend(backend).incr(key, amount, ttl)

    def decr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> int:
        return self.get_backend(backend).decr(key, amount, ttl)

    def ttl(self, key: str, *, backend: str | None = None) -> float | None:
        return self.get_backend(backend).ttl(key)

    def touch(
        self,
        key: str,
        ttl: float,
        *,
        backend: str | None = None,
    ) -> None:
        self.get_backend(backend).touch(key, ttl)

    def is_healthy(self, *, backend: str | None = None) -> bool:
        return self.get_backend(backend).is_healthy()

    def close(self, *, backend: str | None = None) -> None:
        """
        Close either a specific backend or all backends.
        """
        if backend is None:
            for b in self._backends.values():
                with suppress(Exception):
                    b.close()
        else:
            with suppress(Exception):
                self.get_backend(backend).close()

    # --------------------------- Async API --------------------------- #

    async def aget(
        self,
        key: str,
        *,
        backend: str | None = None,
    ) -> Any | None:
        return await self.get_backend(backend).aget(key)

    async def aset(
        self,
        key: str,
        value: Any,
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> None:
        await self.get_backend(backend).aset(key, value, ttl)

    async def adelete(self, key: str, *, backend: str | None = None) -> None:
        await self.get_backend(backend).adelete(key)

    async def aexists(self, key: str, *, backend: str | None = None) -> bool:
        return await self.get_backend(backend).aexists(key)

    async def aclear(self, *, backend: str | None = None) -> None:
        await self.get_backend(backend).aclear()

    async def aget_many(
        self, keys: list[str], *, backend: str | None = None
    ) -> dict[str, Any | None]:
        return await self.get_backend(backend).aget_many(keys)

    async def aset_many(
        self,
        items: dict[str, Any],
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> None:
        await self.get_backend(backend).aset_many(items, ttl)

    async def adelete_many(
        self, keys: list[str], *, backend: str | None = None
    ) -> None:
        await self.get_backend(backend).adelete_many(keys)

    async def aincr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> int:
        return await self.get_backend(backend).aincr(key, amount, ttl)

    async def adecr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
        *,
        backend: str | None = None,
    ) -> int:
        return await self.get_backend(backend).adecr(key, amount, ttl)

    async def attl(
        self,
        key: str,
        *,
        backend: str | None = None,
    ) -> float | None:
        return await self.get_backend(backend).attl(key)

    async def atouch(
        self,
        key: str,
        ttl: float,
        *,
        backend: str | None = None,
    ) -> None:
        await self.get_backend(backend).atouch(key, ttl)

    async def ais_healthy(self, *, backend: str | None = None) -> bool:
        return await self.get_backend(backend).ais_healthy()

    async def aclose(self, *, backend: str | None = None) -> None:
        """
        Close either a specific backend or all backends asynchronously.

        Note: Individual backends may only support sync close; this method
        will best-effort call async close if present,
        or fallback to sync close.
        """
        if backend is None:
            tasks: list[asyncio.Future | asyncio.Task] = []
            for b in self._backends.values():
                if hasattr(b, "aclose"):
                    tasks.append(asyncio.create_task(b.aclose()))
                else:
                    # Fallback to running sync close in a thread if needed
                    loop = asyncio.get_running_loop()
                    tasks.append(loop.run_in_executor(None, b.close))
            if tasks:
                await asyncio.gather(*tasks, return_exceptions=True)
        else:
            b = self.get_backend(backend)
            if hasattr(b, "aclose"):
                await b.aclose()
            else:
                loop = asyncio.get_running_loop()
                await loop.run_in_executor(None, b.close)

    # ---------------------- Context Managers ------------------------ #

    @contextmanager
    def using(
        self,
        backend: str | None = None,
    ) -> Generator[CacheClient, None, None]:
        """
        Context manager yielding a CacheClient bound to a selected backend.

        Example:
            with manager.using("memory") as cache:
                cache.set("k", "v")
                v = cache.get("k")
        """
        client = CacheClient(self, backend)
        try:
            yield client
        finally:
            # No-op; backends are managed by the manager
            pass

    @asynccontextmanager
    async def ausing(
        self, backend: str | None = None
    ) -> AsyncGenerator[AsyncCacheClient, None]:
        """
        Async context manager yielding an AsyncCacheClient
        bound to a selected backend.

        Example:
            async with manager.ausing("redis") as cache:
                await cache.aset("k", "v")
                v = await cache.aget("k")
        """
        client = AsyncCacheClient(self, backend)
        try:
            yield client
        finally:
            # No-op; backends are managed by the manager
            pass

Functions

__init__ 

__init__(config: CacheManagerConfig | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(self, config: CacheManagerConfig | None = None) -> None:
    # Idempotent init to support singleton semantics.
    if getattr(self, "_initialized", False):
        # If new config provided later, allow reconfigure explicitly
        if config is not None:
            self._reconfigure(config)
        return

    self._initialized = True
    self.config = config or CacheManagerConfig()
    self._backends: dict[str, BaseBackend] = {}
    self._default_backend_name: str | None = self.config.default_backend
    self._initialize_backends()

get_backend 

get_backend(name: str | None = None) -> BaseBackend

Retrieve a backend by name, or the default if name is None.

Source code in src/jinpy_utils/cache/core.py 
114
115
116
117
118
119
120
121
122
123
124
125
def get_backend(self, name: str | None = None) -> BaseBackend:
    """
    Retrieve a backend by name, or the default if name is None.
    """
    bname = name or self._default_backend_name
    if not bname or bname not in self._backends:
        raise CacheConfigurationError(
            "Requested backend not found",
            config_section="backend",
            config_value=str(bname),
        )
    return self._backends[bname]

get 

get(key: str, *, backend: str | None = None) -> Any | None
Source code in src/jinpy_utils/cache/core.py 
129
130
def get(self, key: str, *, backend: str | None = None) -> Any | None:
    return self.get_backend(backend).get(key)

set 

set(
    key: str,
    value: Any,
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> None
Source code in src/jinpy_utils/cache/core.py 
132
133
134
135
136
137
138
139
140
def set(
    self,
    key: str,
    value: Any,
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> None:
    self.get_backend(backend).set(key, value, ttl)

delete 

delete(key: str, *, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
142
143
def delete(self, key: str, *, backend: str | None = None) -> None:
    self.get_backend(backend).delete(key)

exists 

exists(key: str, *, backend: str | None = None) -> bool
Source code in src/jinpy_utils/cache/core.py 
145
146
def exists(self, key: str, *, backend: str | None = None) -> bool:
    return self.get_backend(backend).exists(key)

clear 

clear(*, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
148
149
def clear(self, *, backend: str | None = None) -> None:
    self.get_backend(backend).clear()

get_many 

get_many(
    keys: list[str], *, backend: str | None = None
) -> dict[str, Any | None]
Source code in src/jinpy_utils/cache/core.py 
151
152
153
154
def get_many(
    self, keys: list[str], *, backend: str | None = None
) -> dict[str, Any | None]:
    return self.get_backend(backend).get_many(keys)

set_many 

set_many(
    items: dict[str, Any],
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> None
Source code in src/jinpy_utils/cache/core.py 
156
157
158
159
160
161
162
163
def set_many(
    self,
    items: dict[str, Any],
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> None:
    self.get_backend(backend).set_many(items, ttl)

delete_many 

delete_many(keys: list[str], *, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
165
166
167
168
169
170
171
def delete_many(
    self,
    keys: list[str],
    *,
    backend: str | None = None,
) -> None:
    self.get_backend(backend).delete_many(keys)

incr 

incr(
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> int
Source code in src/jinpy_utils/cache/core.py 
173
174
175
176
177
178
179
180
181
def incr(
    self,
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> int:
    return self.get_backend(backend).incr(key, amount, ttl)

decr 

decr(
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> int
Source code in src/jinpy_utils/cache/core.py 
183
184
185
186
187
188
189
190
191
def decr(
    self,
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> int:
    return self.get_backend(backend).decr(key, amount, ttl)

ttl 

ttl(key: str, *, backend: str | None = None) -> float | None
Source code in src/jinpy_utils/cache/core.py 
193
194
def ttl(self, key: str, *, backend: str | None = None) -> float | None:
    return self.get_backend(backend).ttl(key)

touch 

touch(key: str, ttl: float, *, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
196
197
198
199
200
201
202
203
def touch(
    self,
    key: str,
    ttl: float,
    *,
    backend: str | None = None,
) -> None:
    self.get_backend(backend).touch(key, ttl)

is_healthy 

is_healthy(*, backend: str | None = None) -> bool
Source code in src/jinpy_utils/cache/core.py 
205
206
def is_healthy(self, *, backend: str | None = None) -> bool:
    return self.get_backend(backend).is_healthy()

close 

close(*, backend: str | None = None) -> None

Close either a specific backend or all backends.

Source code in src/jinpy_utils/cache/core.py 
208
209
210
211
212
213
214
215
216
217
218
def close(self, *, backend: str | None = None) -> None:
    """
    Close either a specific backend or all backends.
    """
    if backend is None:
        for b in self._backends.values():
            with suppress(Exception):
                b.close()
    else:
        with suppress(Exception):
            self.get_backend(backend).close()

aget async 

aget(key: str, *, backend: str | None = None) -> Any | None
Source code in src/jinpy_utils/cache/core.py 
222
223
224
225
226
227
228
async def aget(
    self,
    key: str,
    *,
    backend: str | None = None,
) -> Any | None:
    return await self.get_backend(backend).aget(key)

aset async 

aset(
    key: str,
    value: Any,
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> None
Source code in src/jinpy_utils/cache/core.py 
230
231
232
233
234
235
236
237
238
async def aset(
    self,
    key: str,
    value: Any,
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> None:
    await self.get_backend(backend).aset(key, value, ttl)

adelete async 

adelete(key: str, *, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
240
241
async def adelete(self, key: str, *, backend: str | None = None) -> None:
    await self.get_backend(backend).adelete(key)

aexists async 

aexists(key: str, *, backend: str | None = None) -> bool
Source code in src/jinpy_utils/cache/core.py 
243
244
async def aexists(self, key: str, *, backend: str | None = None) -> bool:
    return await self.get_backend(backend).aexists(key)

aclear async 

aclear(*, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
246
247
async def aclear(self, *, backend: str | None = None) -> None:
    await self.get_backend(backend).aclear()

aget_many async 

aget_many(
    keys: list[str], *, backend: str | None = None
) -> dict[str, Any | None]
Source code in src/jinpy_utils/cache/core.py 
249
250
251
252
async def aget_many(
    self, keys: list[str], *, backend: str | None = None
) -> dict[str, Any | None]:
    return await self.get_backend(backend).aget_many(keys)

aset_many async 

aset_many(
    items: dict[str, Any],
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> None
Source code in src/jinpy_utils/cache/core.py 
254
255
256
257
258
259
260
261
async def aset_many(
    self,
    items: dict[str, Any],
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> None:
    await self.get_backend(backend).aset_many(items, ttl)

adelete_many async 

adelete_many(keys: list[str], *, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
263
264
265
266
async def adelete_many(
    self, keys: list[str], *, backend: str | None = None
) -> None:
    await self.get_backend(backend).adelete_many(keys)

aincr async 

aincr(
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> int
Source code in src/jinpy_utils/cache/core.py 
268
269
270
271
272
273
274
275
276
async def aincr(
    self,
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> int:
    return await self.get_backend(backend).aincr(key, amount, ttl)

adecr async 

adecr(
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None
) -> int
Source code in src/jinpy_utils/cache/core.py 
278
279
280
281
282
283
284
285
286
async def adecr(
    self,
    key: str,
    amount: int = 1,
    ttl: float | None = None,
    *,
    backend: str | None = None,
) -> int:
    return await self.get_backend(backend).adecr(key, amount, ttl)

attl async 

attl(key: str, *, backend: str | None = None) -> float | None
Source code in src/jinpy_utils/cache/core.py 
288
289
290
291
292
293
294
async def attl(
    self,
    key: str,
    *,
    backend: str | None = None,
) -> float | None:
    return await self.get_backend(backend).attl(key)

atouch async 

atouch(key: str, ttl: float, *, backend: str | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
296
297
298
299
300
301
302
303
async def atouch(
    self,
    key: str,
    ttl: float,
    *,
    backend: str | None = None,
) -> None:
    await self.get_backend(backend).atouch(key, ttl)

ais_healthy async 

ais_healthy(*, backend: str | None = None) -> bool
Source code in src/jinpy_utils/cache/core.py 
305
306
async def ais_healthy(self, *, backend: str | None = None) -> bool:
    return await self.get_backend(backend).ais_healthy()

aclose async 

aclose(*, backend: str | None = None) -> None

Close either a specific backend or all backends asynchronously.

Note: Individual backends may only support sync close; this method will best-effort call async close if present, or fallback to sync close.

Source code in src/jinpy_utils/cache/core.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
async def aclose(self, *, backend: str | None = None) -> None:
    """
    Close either a specific backend or all backends asynchronously.

    Note: Individual backends may only support sync close; this method
    will best-effort call async close if present,
    or fallback to sync close.
    """
    if backend is None:
        tasks: list[asyncio.Future | asyncio.Task] = []
        for b in self._backends.values():
            if hasattr(b, "aclose"):
                tasks.append(asyncio.create_task(b.aclose()))
            else:
                # Fallback to running sync close in a thread if needed
                loop = asyncio.get_running_loop()
                tasks.append(loop.run_in_executor(None, b.close))
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)
    else:
        b = self.get_backend(backend)
        if hasattr(b, "aclose"):
            await b.aclose()
        else:
            loop = asyncio.get_running_loop()
            await loop.run_in_executor(None, b.close)

using 

using(backend: str | None = None) -> Generator[CacheClient, None, None]

Context manager yielding a CacheClient bound to a selected backend.

Example

with manager.using("memory") as cache: cache.set("k", "v") v = cache.get("k")

Source code in src/jinpy_utils/cache/core.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
@contextmanager
def using(
    self,
    backend: str | None = None,
) -> Generator[CacheClient, None, None]:
    """
    Context manager yielding a CacheClient bound to a selected backend.

    Example:
        with manager.using("memory") as cache:
            cache.set("k", "v")
            v = cache.get("k")
    """
    client = CacheClient(self, backend)
    try:
        yield client
    finally:
        # No-op; backends are managed by the manager
        pass

ausing async 

ausing(backend: str | None = None) -> AsyncGenerator[AsyncCacheClient, None]

Async context manager yielding an AsyncCacheClient bound to a selected backend.

Example

async with manager.ausing("redis") as cache: await cache.aset("k", "v") v = await cache.aget("k")

Source code in src/jinpy_utils/cache/core.py 
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
@asynccontextmanager
async def ausing(
    self, backend: str | None = None
) -> AsyncGenerator[AsyncCacheClient, None]:
    """
    Async context manager yielding an AsyncCacheClient
    bound to a selected backend.

    Example:
        async with manager.ausing("redis") as cache:
            await cache.aset("k", "v")
            v = await cache.aget("k")
    """
    client = AsyncCacheClient(self, backend)
    try:
        yield client
    finally:
        # No-op; backends are managed by the manager
        pass

Cache

Cache

High-level cache facade API.

This class delegates to a process-wide CacheManager instance under the hood and provides both sync and async methods for common operations. It is a thin wrapper that preserves the backend selection and promotes simple usage.

Usage (sync): cache = Cache() # uses default manager (singleton) cache.set("k", {"v": 1}, ttl=60) value = cache.get("k")

Usage (async): cache = Cache() await cache.aset("k", "v", ttl=30) v = await cache.aget("k")

Source code in src/jinpy_utils/cache/core.py 
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
class Cache:
    """
    High-level cache facade API.

    This class delegates to a process-wide CacheManager instance under the hood
    and provides both sync and async methods for common operations.
    It is a thin wrapper that preserves
    the backend selection and promotes simple usage.

    Usage (sync):
        cache = Cache()  # uses default manager (singleton)
        cache.set("k", {"v": 1}, ttl=60)
        value = cache.get("k")

    Usage (async):
        cache = Cache()
        await cache.aset("k", "v", ttl=30)
        v = await cache.aget("k")
    """

    def __init__(
        self,
        backend: str | None = None,
        manager: CacheManager | None = None,
    ) -> None:
        self._manager = manager or CacheManager()
        self._backend = backend

    # --------------------------- Sync API --------------------------- #

    def get(self, key: str) -> Any | None:
        return self._manager.get(key, backend=self._backend)

    def set(self, key: str, value: Any, ttl: float | None = None) -> None:
        self._manager.set(key, value, ttl, backend=self._backend)

    def delete(self, key: str) -> None:
        self._manager.delete(key, backend=self._backend)

    def exists(self, key: str) -> bool:
        return self._manager.exists(key, backend=self._backend)

    def clear(self) -> None:
        self._manager.clear(backend=self._backend)

    def get_many(self, keys: list[str]) -> dict[str, Any | None]:
        return self._manager.get_many(keys, backend=self._backend)

    def set_many(
        self,
        items: Mapping[str, Any],
        ttl: float | None = None,
    ) -> None:
        self._manager.set_many(dict(items), ttl, backend=self._backend)

    def delete_many(self, keys: list[str]) -> None:
        self._manager.delete_many(keys, backend=self._backend)

    def incr(self, key: str, amount: int = 1, ttl: float | None = None) -> int:
        return self._manager.incr(key, amount, ttl, backend=self._backend)

    def decr(self, key: str, amount: int = 1, ttl: float | None = None) -> int:
        return self._manager.decr(key, amount, ttl, backend=self._backend)

    def ttl(self, key: str) -> float | None:
        return self._manager.ttl(key, backend=self._backend)

    def touch(self, key: str, ttl: float) -> None:
        self._manager.touch(key, ttl, backend=self._backend)

    def is_healthy(self) -> bool:
        return self._manager.is_healthy(backend=self._backend)

    def close(self) -> None:
        self._manager.close(backend=self._backend)

    # --------------------------- Async API --------------------------- #

    async def aget(self, key: str) -> Any | None:
        return await self._manager.aget(key, backend=self._backend)

    async def aset(
        self,
        key: str,
        value: Any,
        ttl: float | None = None,
    ) -> None:
        await self._manager.aset(key, value, ttl, backend=self._backend)

    async def adelete(self, key: str) -> None:
        await self._manager.adelete(key, backend=self._backend)

    async def aexists(self, key: str) -> bool:
        return await self._manager.aexists(key, backend=self._backend)

    async def aclear(self) -> None:
        await self._manager.aclear(backend=self._backend)

    async def aget_many(self, keys: list[str]) -> dict[str, Any | None]:
        return await self._manager.aget_many(keys, backend=self._backend)

    async def aset_many(
        self, items: Mapping[str, Any], ttl: float | None = None
    ) -> None:
        await self._manager.aset_many(dict(items), ttl, backend=self._backend)

    async def adelete_many(self, keys: list[str]) -> None:
        await self._manager.adelete_many(keys, backend=self._backend)

    async def aincr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
    ) -> int:
        return await self._manager.aincr(
            key,
            amount,
            ttl,
            backend=self._backend,
        )

    async def adecr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
    ) -> int:
        return await self._manager.adecr(
            key,
            amount,
            ttl,
            backend=self._backend,
        )

    async def attl(self, key: str) -> float | None:
        return await self._manager.attl(key, backend=self._backend)

    async def atouch(self, key: str, ttl: float) -> None:
        await self._manager.atouch(key, ttl, backend=self._backend)

    async def ais_healthy(self) -> bool:
        return await self._manager.ais_healthy(backend=self._backend)

    async def aclose(self) -> None:
        await self._manager.aclose(backend=self._backend)

Functions

__init__ 

__init__(
    backend: str | None = None, manager: CacheManager | None = None
) -> None
Source code in src/jinpy_utils/cache/core.py 
529
530
531
532
533
534
535
def __init__(
    self,
    backend: str | None = None,
    manager: CacheManager | None = None,
) -> None:
    self._manager = manager or CacheManager()
    self._backend = backend

get 

get(key: str) -> Any | None
Source code in src/jinpy_utils/cache/core.py 
539
540
def get(self, key: str) -> Any | None:
    return self._manager.get(key, backend=self._backend)

set 

set(key: str, value: Any, ttl: float | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
542
543
def set(self, key: str, value: Any, ttl: float | None = None) -> None:
    self._manager.set(key, value, ttl, backend=self._backend)

delete 

delete(key: str) -> None
Source code in src/jinpy_utils/cache/core.py 
545
546
def delete(self, key: str) -> None:
    self._manager.delete(key, backend=self._backend)

exists 

exists(key: str) -> bool
Source code in src/jinpy_utils/cache/core.py 
548
549
def exists(self, key: str) -> bool:
    return self._manager.exists(key, backend=self._backend)

clear 

clear() -> None
Source code in src/jinpy_utils/cache/core.py 
551
552
def clear(self) -> None:
    self._manager.clear(backend=self._backend)

get_many 

get_many(keys: list[str]) -> dict[str, Any | None]
Source code in src/jinpy_utils/cache/core.py 
554
555
def get_many(self, keys: list[str]) -> dict[str, Any | None]:
    return self._manager.get_many(keys, backend=self._backend)

set_many 

set_many(items: Mapping[str, Any], ttl: float | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
557
558
559
560
561
562
def set_many(
    self,
    items: Mapping[str, Any],
    ttl: float | None = None,
) -> None:
    self._manager.set_many(dict(items), ttl, backend=self._backend)

delete_many 

delete_many(keys: list[str]) -> None
Source code in src/jinpy_utils/cache/core.py 
564
565
def delete_many(self, keys: list[str]) -> None:
    self._manager.delete_many(keys, backend=self._backend)

incr 

incr(key: str, amount: int = 1, ttl: float | None = None) -> int
Source code in src/jinpy_utils/cache/core.py 
567
568
def incr(self, key: str, amount: int = 1, ttl: float | None = None) -> int:
    return self._manager.incr(key, amount, ttl, backend=self._backend)

decr 

decr(key: str, amount: int = 1, ttl: float | None = None) -> int
Source code in src/jinpy_utils/cache/core.py 
570
571
def decr(self, key: str, amount: int = 1, ttl: float | None = None) -> int:
    return self._manager.decr(key, amount, ttl, backend=self._backend)

ttl 

ttl(key: str) -> float | None
Source code in src/jinpy_utils/cache/core.py 
573
574
def ttl(self, key: str) -> float | None:
    return self._manager.ttl(key, backend=self._backend)

touch 

touch(key: str, ttl: float) -> None
Source code in src/jinpy_utils/cache/core.py 
576
577
def touch(self, key: str, ttl: float) -> None:
    self._manager.touch(key, ttl, backend=self._backend)

is_healthy 

is_healthy() -> bool
Source code in src/jinpy_utils/cache/core.py 
579
580
def is_healthy(self) -> bool:
    return self._manager.is_healthy(backend=self._backend)

close 

close() -> None
Source code in src/jinpy_utils/cache/core.py 
582
583
def close(self) -> None:
    self._manager.close(backend=self._backend)

aget async 

aget(key: str) -> Any | None
Source code in src/jinpy_utils/cache/core.py 
587
588
async def aget(self, key: str) -> Any | None:
    return await self._manager.aget(key, backend=self._backend)

aset async 

aset(key: str, value: Any, ttl: float | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
590
591
592
593
594
595
596
async def aset(
    self,
    key: str,
    value: Any,
    ttl: float | None = None,
) -> None:
    await self._manager.aset(key, value, ttl, backend=self._backend)

adelete async 

adelete(key: str) -> None
Source code in src/jinpy_utils/cache/core.py 
598
599
async def adelete(self, key: str) -> None:
    await self._manager.adelete(key, backend=self._backend)

aexists async 

aexists(key: str) -> bool
Source code in src/jinpy_utils/cache/core.py 
601
602
async def aexists(self, key: str) -> bool:
    return await self._manager.aexists(key, backend=self._backend)

aclear async 

aclear() -> None
Source code in src/jinpy_utils/cache/core.py 
604
605
async def aclear(self) -> None:
    await self._manager.aclear(backend=self._backend)

aget_many async 

aget_many(keys: list[str]) -> dict[str, Any | None]
Source code in src/jinpy_utils/cache/core.py 
607
608
async def aget_many(self, keys: list[str]) -> dict[str, Any | None]:
    return await self._manager.aget_many(keys, backend=self._backend)

aset_many async 

aset_many(items: Mapping[str, Any], ttl: float | None = None) -> None
Source code in src/jinpy_utils/cache/core.py 
610
611
612
613
async def aset_many(
    self, items: Mapping[str, Any], ttl: float | None = None
) -> None:
    await self._manager.aset_many(dict(items), ttl, backend=self._backend)

adelete_many async 

adelete_many(keys: list[str]) -> None
Source code in src/jinpy_utils/cache/core.py 
615
616
async def adelete_many(self, keys: list[str]) -> None:
    await self._manager.adelete_many(keys, backend=self._backend)

aincr async 

aincr(key: str, amount: int = 1, ttl: float | None = None) -> int
Source code in src/jinpy_utils/cache/core.py 
618
619
620
621
622
623
624
625
626
627
628
629
async def aincr(
    self,
    key: str,
    amount: int = 1,
    ttl: float | None = None,
) -> int:
    return await self._manager.aincr(
        key,
        amount,
        ttl,
        backend=self._backend,
    )

adecr async 

adecr(key: str, amount: int = 1, ttl: float | None = None) -> int
Source code in src/jinpy_utils/cache/core.py 
631
632
633
634
635
636
637
638
639
640
641
642
async def adecr(
    self,
    key: str,
    amount: int = 1,
    ttl: float | None = None,
) -> int:
    return await self._manager.adecr(
        key,
        amount,
        ttl,
        backend=self._backend,
    )

attl async 

attl(key: str) -> float | None
Source code in src/jinpy_utils/cache/core.py 
644
645
async def attl(self, key: str) -> float | None:
    return await self._manager.attl(key, backend=self._backend)

atouch async 

atouch(key: str, ttl: float) -> None
Source code in src/jinpy_utils/cache/core.py 
647
648
async def atouch(self, key: str, ttl: float) -> None:
    await self._manager.atouch(key, ttl, backend=self._backend)

ais_healthy async 

ais_healthy() -> bool
Source code in src/jinpy_utils/cache/core.py 
650
651
async def ais_healthy(self) -> bool:
    return await self._manager.ais_healthy(backend=self._backend)

aclose async 

aclose() -> None
Source code in src/jinpy_utils/cache/core.py 
653
654
async def aclose(self) -> None:
    await self._manager.aclose(backend=self._backend)

CacheClient

CacheClient

Thin synchronous facade over CacheManager bound to a specific backend.

Source code in src/jinpy_utils/cache/core.py 
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
class CacheClient:
    """
    Thin synchronous facade over CacheManager bound to a specific backend.
    """

    def __init__(self, manager: CacheManager, backend: str | None) -> None:
        self._manager = manager
        self._backend = backend

    def get(self, key: str) -> Any | None:
        return self._manager.get(key, backend=self._backend)

    def set(self, key: str, value: Any, ttl: float | None = None) -> None:
        self._manager.set(key, value, ttl, backend=self._backend)

    def delete(self, key: str) -> None:
        self._manager.delete(key, backend=self._backend)

    def exists(self, key: str) -> bool:
        return self._manager.exists(key, backend=self._backend)

    def clear(self) -> None:
        self._manager.clear(backend=self._backend)

    def get_many(self, keys: list[str]) -> dict[str, Any | None]:
        return self._manager.get_many(keys, backend=self._backend)

    def set_many(
        self,
        items: dict[str, Any],
        ttl: float | None = None,
    ) -> None:
        self._manager.set_many(items, ttl, backend=self._backend)

    def delete_many(self, keys: list[str]) -> None:
        self._manager.delete_many(keys, backend=self._backend)

    def incr(self, key: str, amount: int = 1, ttl: float | None = None) -> int:
        return self._manager.incr(key, amount, ttl, backend=self._backend)

    def decr(self, key: str, amount: int = 1, ttl: float | None = None) -> int:
        return self._manager.decr(key, amount, ttl, backend=self._backend)

    def ttl(self, key: str) -> float | None:
        return self._manager.ttl(key, backend=self._backend)

    def touch(self, key: str, ttl: float) -> None:
        self._manager.touch(key, ttl, backend=self._backend)

    def is_healthy(self) -> bool:
        return self._manager.is_healthy(backend=self._backend)

Functions

__init__ 

__init__(manager: CacheManager, backend: str | None) -> None
Source code in src/jinpy_utils/cache/core.py 
383
384
385
def __init__(self, manager: CacheManager, backend: str | None) -> None:
    self._manager = manager
    self._backend = backend

AsyncCacheClient

AsyncCacheClient

Thin asynchronous facade over CacheManager bound to a specific backend.

Source code in src/jinpy_utils/cache/core.py 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
class AsyncCacheClient:
    """
    Thin asynchronous facade over CacheManager bound to a specific backend.
    """

    def __init__(self, manager: CacheManager, backend: str | None) -> None:
        self._manager = manager
        self._backend = backend

    async def aget(self, key: str) -> Any | None:
        return await self._manager.aget(key, backend=self._backend)

    async def aset(
        self,
        key: str,
        value: Any,
        ttl: float | None = None,
    ) -> None:
        await self._manager.aset(key, value, ttl, backend=self._backend)

    async def adelete(self, key: str) -> None:
        await self._manager.adelete(key, backend=self._backend)

    async def aexists(self, key: str) -> bool:
        return await self._manager.aexists(key, backend=self._backend)

    async def aclear(self) -> None:
        await self._manager.aclear(backend=self._backend)

    async def aget_many(self, keys: list[str]) -> dict[str, Any | None]:
        return await self._manager.aget_many(keys, backend=self._backend)

    async def aset_many(
        self,
        items: dict[str, Any],
        ttl: float | None = None,
    ) -> None:
        await self._manager.aset_many(items, ttl, backend=self._backend)

    async def adelete_many(self, keys: list[str]) -> None:
        await self._manager.adelete_many(keys, backend=self._backend)

    async def aincr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
    ) -> int:
        return await self._manager.aincr(
            key,
            amount,
            ttl,
            backend=self._backend,
        )

    async def adecr(
        self,
        key: str,
        amount: int = 1,
        ttl: float | None = None,
    ) -> int:
        return await self._manager.adecr(
            key,
            amount,
            ttl,
            backend=self._backend,
        )

    async def attl(self, key: str) -> float | None:
        return await self._manager.attl(key, backend=self._backend)

    async def atouch(self, key: str, ttl: float) -> None:
        await self._manager.atouch(key, ttl, backend=self._backend)

    async def ais_healthy(self) -> bool:
        return await self._manager.ais_healthy(backend=self._backend)

Functions

__init__ 

__init__(manager: CacheManager, backend: str | None) -> None
Source code in src/jinpy_utils/cache/core.py 
436
437
438
def __init__(self, manager: CacheManager, backend: str | None) -> None:
    self._manager = manager
    self._backend = backend

Interfaces

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: ...

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: ...

Utilities

normalize_key

normalize_key 

normalize_key(key: str) -> str

Normalize cache key to a safe form.

Source code in src/jinpy_utils/cache/utils.py 
14
15
16
17
18
19
20
def normalize_key(key: str) -> str:
    """Normalize cache key to a safe form."""
    key = key.strip()
    if not key:
        raise ValueError("Cache key cannot be empty")
    # Prevent accidental whitespace/control characters
    return "".join(ch for ch in key if ch.isprintable())

compute_expiry

compute_expiry 

compute_expiry(ttl: float | None) -> float | None

Compute absolute expiry timestamp in seconds.

Source code in src/jinpy_utils/cache/utils.py 
28
29
30
31
32
33
34
35
def compute_expiry(ttl: float | None) -> float | None:
    """Compute absolute expiry timestamp in seconds."""
    if ttl is None:
        return None
    if ttl <= 0:
        # Immediate expiry
        return now_seconds() - 1
    return now_seconds() + ttl

remaining_ttl

remaining_ttl 

remaining_ttl(expiry: float | None) -> float | None

Compute remaining ttl from absolute expiry.

Source code in src/jinpy_utils/cache/utils.py 
38
39
40
41
42
43
def remaining_ttl(expiry: float | None) -> float | None:
    """Compute remaining ttl from absolute expiry."""
    if expiry is None:
        return None
    remaining = expiry - now_seconds()
    return remaining if remaining > 0 else 0.0

default_serializer

default_serializer 

default_serializer(
    kind: SerializerType,
) -> tuple[Callable[[Any], bytes], Callable[[bytes], Any]]

Return serializer, deserializer functions based on kind.

Source code in src/jinpy_utils/cache/utils.py 
46
47
48
49
50
51
52
53
54
55
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
def default_serializer(
    kind: SerializerType,
) -> tuple[Callable[[Any], bytes], Callable[[bytes], Any]]:
    """Return serializer, deserializer functions based on kind."""
    if kind == "json":

        def ser(obj: Any) -> bytes:
            return json.dumps(
                obj, default=str, separators=(",", ":"), ensure_ascii=False
            ).encode("utf-8")

        def de(data: bytes) -> Any:
            return json.loads(data.decode("utf-8"))

        return ser, de

    if kind == "pickle":

        def ser(obj: Any) -> bytes:
            return pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)

        def de(data: bytes) -> Any:
            return pickle.loads(data)

        return ser, de

    if kind == "str":

        def ser(obj: Any) -> bytes:
            return str(obj).encode("utf-8")

        def de(data: bytes) -> Any:
            return data.decode("utf-8")

        return ser, de

    if kind == "bytes":

        def ser(obj: Any) -> bytes:
            if isinstance(obj, (bytes | bytearray | memoryview)):
                return bytes(obj)
            raise TypeError("Expected bytes-like value")

        def de(data: bytes) -> Any:
            return data

        return ser, de

    raise ValueError(f"Unsupported serializer type: {kind}")

Examples

Basic Usage (in-memory)

from jinpy_utils.cache import get_cache

cache = get_cache()  # default in-memory backend
cache.set("user:123", {"name": "Ada"}, ttl=60)
print(cache.get("user:123"))  # {"name": "Ada"}

Async Usage

import asyncio
from jinpy_utils.cache import get_cache

async def main():
    cache = get_cache()
    await cache.aset("token", "abc", ttl=10)
    value = await cache.aget("token")
    print(value)

asyncio.run(main())

Using a context client bound to a backend

from jinpy_utils.cache import CacheManager, MemoryCacheConfig

manager = CacheManager()
# Ensure a named memory backend exists (defaults will be auto-provisioned)
with manager.using("default_memory") as c:
    c.set("k", 1)
    assert c.incr("k") == 2

Reconfiguring the manager with multiple backends

from pathlib import Path
from jinpy_utils.cache import (
    CacheManager, CacheManagerConfig,
    MemoryCacheConfig, FileCacheConfig
)

config = CacheManagerConfig(
    default_backend="mem",
    backends=[
        MemoryCacheConfig(name="mem", default_ttl=120),
        FileCacheConfig(name="disk", directory=Path(".cache"))
    ]
)

manager = CacheManager(config)  # initializes both

# Use the file backend via high-level facade
from jinpy_utils.cache import Cache
file_cache = Cache(backend="disk")
file_cache.set("report", {"ok": True})

Atomic counters with TTL

from jinpy_utils.cache import get_cache

cache = get_cache()
count = cache.incr("jobs:completed", amount=1, ttl=300)