Documentation Index
Fetch the complete documentation index at: https://mintlify.com/Indroic/HexCore/llms.txt
Use this file to discover all available pages before exploring further.
HexCore ships a lightweight, pluggable caching abstraction — ICache — with two ready-to-use backends: MemoryCache for development and single-process deployments, and RedisCache for production scenarios that need shared state across multiple workers or replicas. This page covers the interface, both implementations, configuration, and usage patterns in domain services and use cases.
The ICache Interface
All cache backends implement the abstract class defined in hexcore.infrastructure.cache:
import abc
import typing as t
class ICache(abc.ABC):
@abc.abstractmethod
async def get(self, key: str) -> t.Optional[t.Any]:
"""Return the cached value, or None if the key does not exist."""
@abc.abstractmethod
def set(self, key: str, value: t.Any, expire: int = 3600) -> t.Any:
"""Store value under key with an optional TTL in seconds."""
@abc.abstractmethod
def delete(self, key: str) -> t.Union[t.Any, None]:
"""Remove the key from the cache."""
get
async (key: str) -> Optional[Any]
Retrieves a cached value by key. Returns None on a cache miss.
set
(key: str, value: Any, expire: int = 3600) -> Any
Stores a value under the given key. expire is the TTL in seconds (default 3600). The MemoryCache ignores the TTL; RedisCache passes it to Redis as EX.
Removes a key from the cache. A no-op if the key does not exist.
Built-in Backends
MemoryCache stores data in a plain Python dictionary. It is zero-dependency and requires no external services, making it ideal for unit tests, local development, or single-process applications.from hexcore.infrastructure.cache.cache_backends.memory import MemoryCache
cache = MemoryCache()
await cache.set("user:42", {"name": "Alice"})
value = await cache.get("user:42") # {"name": "Alice"}
await cache.delete("user:42")
await cache.clear() # wipe entire cache
MemoryCache data lives in process memory and is not shared between workers or replicas. Do not use it in horizontally-scaled deployments where cache consistency matters.
Implementation details:
- Internal store:
dict[str, dict[str, Any]]
expire parameter accepted but not enforced — keys never expire automatically
- Thread-safe for single-threaded asyncio event loops
- Extra
clear() method to flush all keys at once
RedisCache connects to Redis using redis.asyncio and serialises values as JSON. It reads redis_uri and redis_cache_duration from ServerConfig via LazyConfig.from hexcore.infrastructure.cache.cache_backends.redis import RedisCache
cache = RedisCache() # auto-configures from ServerConfig
await cache.set("user:42", {"name": "Alice"}, expire=600)
value = await cache.get("user:42") # {"name": "Alice"} or None
await cache.delete("user:42")
await cache.clear() # FLUSHDB — use with care in production
Installation:pip install "hexcore[redis]"
Configuration:from hexcore.config import ServerConfig
config = ServerConfig(
redis_uri="redis://localhost:6379/0",
redis_cache_duration=3600, # default TTL in seconds
)
Implementation details:
- Connects via
redis.Redis.from_url(config.redis_uri, decode_responses=True)
- Values are JSON-serialised on write and deserialised on read
expire defaults to ServerConfig.redis_cache_duration when not specified
clear() calls FLUSHDB — this wipes the entire Redis database, not just HexCore keys
Configuring the Cache Backend
Set the cache_backend field in your ServerConfig to the desired ICache instance:
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=3600,
cache_backend=RedisCache(),
)
For development or testing, swap in MemoryCache:
from hexcore.config import ServerConfig
from hexcore.infrastructure.cache.cache_backends.memory import MemoryCache
config = ServerConfig(
cache_backend=MemoryCache(),
)
Accessing the Cache
Retrieve the configured backend anywhere through LazyConfig:
from hexcore.config import LazyConfig
cache = LazyConfig.get_config().cache_backend
In a Domain Service
src/domain/products/services.py
from hexcore.config import LazyConfig
from src.domain.products.entities import Product
from src.domain.products.repositories import IProductRepository
class ProductCatalogService:
def __init__(self, repo: IProductRepository) -> None:
self.repo = repo
self._cache = LazyConfig.get_config().cache_backend
async def get_product(self, product_id: str) -> Product:
cache_key = f"product:{product_id}"
cached = await self._cache.get(cache_key)
if cached:
return Product(**cached)
product = await self.repo.get_by_id(product_id)
await self._cache.set(cache_key, product.model_dump(), expire=300)
return product
async def invalidate_product(self, product_id: str) -> None:
await self._cache.delete(f"product:{product_id}")
In a Use Case
src/application/use_cases/get_user.py
from hexcore.config import LazyConfig
from src.domain.users.entities import User
from src.domain.users.repositories import IUserRepository
class GetUserUseCase:
def __init__(self, repo: IUserRepository) -> None:
self.repo = repo
async def execute(self, user_id: str) -> User:
cache = LazyConfig.get_config().cache_backend
key = f"user:{user_id}"
hit = await cache.get(key)
if hit:
return User(**hit)
user = await self.repo.get_by_id(user_id)
await cache.set(key, user.model_dump())
return user
Implementing a Custom Backend
You can add any cache store (Memcached, DynamoDB, etc.) by subclassing ICache:
src/infrastructure/cache/custom_cache.py
from hexcore.infrastructure.cache import ICache
import typing as t
class NullCache(ICache):
"""A no-op cache useful in tests where caching must be bypassed."""
async def get(self, key: str) -> t.Optional[t.Any]:
return None
async def set(self, key: str, value: t.Any, expire: int = 3600) -> None:
pass
async def delete(self, key: str) -> None:
pass
Then register it in ServerConfig:
from hexcore.config import ServerConfig
from src.infrastructure.cache.custom_cache import NullCache
config = ServerConfig(cache_backend=NullCache())