CacheRedisAdapter
What it is
- A Redis-backed implementation of
ICacheAdapter. - Stores cache entries as JSON-serialized
CachedDatain Redis. - Uses namespaced, hashed Redis keys:
{prefix}:{sha256(logical_key)}.
Public API
class CacheRedisAdapter(ICacheAdapter)__init__(redis_url: str, prefix: str = "naas:cache", socket_timeout: float | None = None)- Connects to Redis using
redis.Redis.from_url(...). - Normalizes
prefixby stripping trailing:.
- Connects to Redis using
get(key: str) -> CachedData- Fetches and deserializes a
CachedDataentry. - Raises
CacheNotFoundErrorif missing.
- Fetches and deserializes a
set(key: str, value: CachedData) -> None- Stores a
CachedDataentry (JSON).
- Stores a
set_if_absent(key: str, value: CachedData) -> bool- Atomically sets only if key does not exist (Redis
NX). - Returns
Trueif set,Falseotherwise.
- Atomically sets only if key does not exist (Redis
delete(key: str) -> None- Deletes an entry.
- Raises
CacheNotFoundErrorif missing.
exists(key: str) -> bool- Checks existence of an entry.
Configuration/Dependencies
- Requires the
redisPython package (import redis as redis_lib). - Redis connection:
redis_url: e.g.redis://localhost:6379/0socket_timeout: forwarded to the Redis client
- Uses
decode_responses=True(Redis values are returned asstr).
Usage
from naas_abi_core.services.cache.adapters.secondary.CacheRedisAdapter import CacheRedisAdapter
from naas_abi_core.services.cache.CachePort import CachedData
cache = CacheRedisAdapter(redis_url="redis://localhost:6379/0", prefix="naas:cache")
key = "user:123"
value = CachedData(...) # construct per your project's CachedData schema
cache.set(key, value)
if cache.exists(key):
loaded = cache.get(key)
print(loaded)
was_set = cache.set_if_absent(key, value)
print("set_if_absent:", was_set)
cache.delete(key)
Caveats
- Redis keys are not the raw logical keys; they are SHA-256 hashes with a prefix. You cannot easily inspect Redis to find entries by logical key without hashing.
get()anddelete()raiseCacheNotFoundErrorwhen the entry does not exist.