ICache — Pluggable Async Cache Backend Interface

ICache is HexCore’s abstract cache port. Application code depends only on the interface; the concrete backend (in-memory or Redis) is injected through ServerConfig.cache_backend. Swapping from MemoryCache to RedisCache in production requires changing a single line in your config file.

`ICache`

hexcore.infrastructure.cache.ICache Abstract base class that all cache backends must implement.

import abc
import typing as t

class ICache(abc.ABC):
    @abc.abstractmethod
    async def get(self, key: str) -> t.Optional[t.Any]: ...

    @abc.abstractmethod
    def set(self, key: str, value: t.Any, expire: int = 3600) -> t.Any: ...

    @abc.abstractmethod
    def delete(self, key: str) -> t.Union[t.Any, None]: ...

Methods

get(key: str)

Awaitable[Optional[Any]]

Retrieves a cached value by key. Returns None if the key does not exist or has expired.

set(key: str, value: Any, expire: int = 3600)

Any

Stores a value under key. The expire argument is the TTL in seconds; semantics vary by backend (see below).

delete(key: str)

Any | None

Removes the entry for key. No-ops silently if the key does not exist.

`MemoryCache`

hexcore.infrastructure.cache.cache_backends.memory.MemoryCache Simple in-process dictionary cache with no external dependencies. Suitable for development, testing, or single-replica deployments where cache eviction is not required.

from hexcore.infrastructure.cache.cache_backends.memory import MemoryCache

cache = MemoryCache()
await cache.set("user:1", {"name": "Alice"})
value = await cache.get("user:1")   # {"name": "Alice"}
await cache.delete("user:1")
await cache.clear()                 # flush everything

Constructor

MemoryCache()

No parameters. Initialises an empty dict as the backing store.

Methods

Method	Signature	Notes
`get`	`async (key: str) -> Optional[Dict[str, Any]]`	Returns `None` for missing keys
`set`	`async (key: str, value: Dict[str, Any], expire: int = 0) -> None`	`expire` is accepted but not enforced — values never expire
`delete`	`async (key: str) -> None`	Silently no-ops for missing keys
`clear`	`async () -> None`	Flushes the entire in-memory store

MemoryCache does not implement TTL-based expiration. The expire parameter is accepted for interface compatibility but has no effect. Use RedisCache when you need real expiry.

`RedisCache`

hexcore.infrastructure.cache.cache_backends.redis.RedisCache Production-grade cache backend backed by Redis. Values are JSON-serialised before storage and deserialised on retrieval.

Requires the hexcore[redis] or hexcore[cache] extra (installs redis[asyncio]):

pip install "hexcore[redis]"

Constructor

RedisCache()

No parameters. The Redis connection is created lazily from LazyConfig.get_config().redis_uri at instantiation time using redis.asyncio.Redis.from_url(..., decode_responses=True). Configure the URI in your ServerConfig:

config.py

from hexcore.config import ServerConfig

config = ServerConfig(
    redis_uri="redis://my-redis-host:6379/1",
    redis_cache_duration=600,  # seconds — default TTL for set()
)

Methods

Method	Signature	Notes
`get`	`async (key: str) -> Optional[Dict[str, Any]]`	Deserialises JSON; returns `None` for missing/expired keys
`set`	`async (key: str, value: Dict[str, Any], expire: int = <redis_cache_duration>) -> None`	Serialises to JSON; default TTL from `ServerConfig.redis_cache_duration` (300 s)
`delete`	`async (key: str) -> None`	Issues `DEL key`
`clear`	`async () -> None`	Issues `FLUSHDB` — clears the entire Redis database

clear() calls FLUSHDB, which removes all keys in the selected Redis database—not just those written by HexCore. Use with caution in shared Redis instances.

Swapping backends

The cache backend is a first-class field on ServerConfig. Change it once; every service that uses LazyConfig.get_config().cache_backend picks up the new backend automatically.

MemoryCache (default)
RedisCache (production)
Custom backend

config.py

from hexcore.config import ServerConfig
from hexcore.infrastructure.cache.cache_backends.memory import MemoryCache

config = ServerConfig(
    cache_backend=MemoryCache(),
)

config.py

from hexcore.config import ServerConfig
from hexcore.infrastructure.cache.cache_backends.redis import RedisCache

config = ServerConfig(
    redis_uri="redis://localhost:6379/0",
    redis_cache_duration=300,
    cache_backend=RedisCache(),
)

config.py

from hexcore.config import ServerConfig
from hexcore.infrastructure.cache import ICache
import typing as t

class MyCustomCache(ICache):
    async def get(self, key: str) -> t.Optional[t.Any]:
        ...

    def set(self, key: str, value: t.Any, expire: int = 3600) -> t.Any:
        ...

    def delete(self, key: str) -> t.Any | None:
        ...

config = ServerConfig(
    cache_backend=MyCustomCache(),
)

Using the cache in application code

Always resolve the backend through LazyConfig so that tests and different environments can inject their own backend:

from hexcore.config import LazyConfig

async def get_cached_user(user_id: str) -> dict | None:
    cache = LazyConfig.get_config().cache_backend
    key = f"user:{user_id}"

    cached = await cache.get(key)
    if cached:
        return cached

    user = await fetch_user_from_db(user_id)
    await cache.set(key, user, expire=120)
    return user

Domain

CQRS

Infrastructure

ICache — Pluggable Async Cache Backend Interface

`ICache`

Methods

`MemoryCache`

Constructor

Methods

`RedisCache`

Constructor

Methods

Swapping backends

Using the cache in application code

Build docs developers (and LLMs) love

Domain

CQRS

Infrastructure

Documentation Index

​ICache

​Methods

​MemoryCache

​Constructor

​Methods

​RedisCache

​Constructor

​Methods

​Swapping backends

​Using the cache in application code

Build docs developers (and LLMs) love

`ICache`

Methods

`MemoryCache`

Constructor

Methods

`RedisCache`

Constructor

Methods

Swapping backends

Using the cache in application code