Backends — Aquilia Documentation
Comprehensive guide and documentation for Backends in the Aquilia framework. View API reference, examples, and implementation patterns.
Cache / Backends Cache Backends Aquilia ships with four built-in cache backends conforming to the CacheBackend interface. All can be transparently swapped behind the same CacheService instance. Backend Comparison Backend Persistence Distributed Primary Strength Best For ))} CacheBackend Contract class CacheBackend(ABC): async def initialize(self) -> None: ... async def shutdown(self) -> None: ... async def get(self, key: str) -> CacheEntry | None: ... async def set(self, key: str, value: Any, ttl: int | None = None, tags: tuple[str, ...] = (), namespace: str = "default") -> None: ... async def delete(self, key: str) -> bool: ... async def exists(self, key: str) -> bool: ... async def clear(self, namespace: str | None = None) -> int: ... async def keys(self, pattern: str = "*", namespace: str | None = None) -> list[str]: ... async def stats(self) -> CacheStats: ... # Optional methods with defaults in base class: async def delete_by_tags(self, tags: set[str]) -> int: ... async def get_many(self, keys: list[str]) -> dict[str, CacheEntry | None]: ... async def set_many(self, items: dict[str, Any], ttl: int | None = None, namespace: str = "default") -> None: ... async def delete_many(self, keys: list[str]) -> int: ... async def increment(self, key: str, delta: int = 1) -> int | None: ... async def decrement(self, key: str, delta: int = 1) -> int | None: ... @property def name(self) -> str: ... @property def is_distributed(self) -> bool: ... MemoryBackend In-process cache utilizing locks for concurrency control, an asynchronous sweeper task for TTL management, inverted indexes for tag-based groupings, and configurable eviction policies (lru, lfu, fifo, ttl, random). from aquilia.cache import MemoryBackend, CacheService backend = MemoryBackend( max_size=10_000, eviction_policy="lru", sweep_interval=30.0, max_memory_bytes=0, capacity_warning_threshold=0.85, ) cache = CacheService(backend=backend) await cache.initialize() • Stores detailed metadata for each entry, including creation times, access frequencies, namespaces, and sizes. • The active background sweeper runs periodically to evict expired items from a sorted TTL min-heap. RedisBackend A distributed caching backend communicating with Redis databases. Manages connection pooling, pipelined batch transactions, and uses Redis Sets for O(1) tag and namespace lookups. from aquilia.cache import RedisBackend, CacheService backend = RedisBackend( url="redis://localhost:6379/0", max_connections=20, key_prefix="aq:", socket_timeout=5.0, connect_timeout=5.0, retry_on_timeout=True, ) cache = CacheService(backend=backend) await cache.initialize() • Uses namespaces and key prefix mapping. Tags are stored in Redis Sets named _tags: . • Pluggable serialization supports JSON, Msgpack, and Pickle options. CompositeBackend (L1/L2) Combines a fast local L1 memory cache and a distributed L2 Redis cache. Reads hit L1 first, falling back to L2. Found L2 values are promoted to L1, and writes update both levels simultaneously. from aquilia.cache import ( CompositeBackend, MemoryBackend, RedisBackend, CacheService, ) l1 = MemoryBackend(max_size=1_000) l2 = RedisBackend(url="redis://localhost:6379/0") backend = CompositeBackend( l1=l1, l2=l2, promote_on_l2_hit=True, async_l2_write=False, ) cache = CacheService(backend=backend) NullBackend No-op cache backend implementation. Always misses reads and ignores write operations. Useful for testing or disabling cache systems in production without changing application code. from aquilia.cache import NullBackend, CacheService cache = CacheService(backend=NullBackend()) Cache Serializers Serialization formats are crucial for L2/Redis backends. Three serializers are supported: JsonCacheSerializer Default option. Converts values into JSON bytes. Highly portable and secure, but limited to JSON-compatible data types. MsgpackCacheSerializer High-speed, compact binary representation. Requires the optional msgpack package. PickleCacheSerializer Standard Python pickle serialization. Supports arbitrary Python objects. Encrypted/signed using an HMAC signature via secret_key to prevent tampering. from aquilia.cache.serializers import get_serializer json_ser = get_serializer("json") msgpack_ser = get_serializer("msgpack") pickle_ser = get_serializer("pickle", secret_key="secure-hmac-key") Key Builders Key builders normalize logical keys into unique string descriptors. from aquilia.cache import DefaultKeyBuilder, HashKeyBuilder default_builder = DefaultKeyBuilder(version=1) print(default_builder.build(namespace="users", key="42", prefix="aq:")) # Output: aq:v1:users:42 hash_builder = HashKeyBuilder(hash_length=16, version=1) print(hash_builder.build(namespace="search", key="long-query", prefix="aq:")) # Output: aq:v1:search: )
Go to Homepage