Cache Effect — Aquilia Documentation
Comprehensive guide and documentation for Cache Effect in the Aquilia framework. View API reference, examples, and implementation patterns.
EFFECTS / CACHE EFFECT Cache Effect The CacheEffect enables scoped key-value caching in Aquilia request handlers. Managed by the CacheProvider , it partitions cache keys by namespace to prevent key collisions across modules. Namespace Isolation Rather than mixing keys in a single flat database namespace (which leads to key collisions and complex prefixing logic), the CacheEffect partitions operations. By requesting a namespace at initialization (e.g. CacheEffect("products")), the handle transparently prefixes all operations, keeping caching logic localized. Resource Handles Depending on the system state, the CacheProvider yields one of two resource handles: CacheServiceHandle Acquired when the caching integration is enabled. It forwards operations to a central, DI-injected CacheService connected to Redis or Memcached. CacheHandle A fallback handle backed by a standard Python dictionary. Used during unit testing or when the main cache backend is unconfigured, ensuring handlers run without throwing connection exceptions. API Reference Both handles expose the same core async methods, making tests consistent with production: await handle.get(key: str) -> Any | None Retrieves the deserialized cache value. Returns None on a cache miss. await handle.set(key: str, value: Any, ttl: int | None = None) -> None Stores the value inside the namespace. Accepts an optional TTL (in seconds) that defaults to the config value if omitted. await handle.delete(key: str) -> bool Deletes the value matching the key from the namespace. Returns True if the key existed and was deleted, False otherwise. Usage: Cache-Aside with Automatic Invalidation The example below demonstrates retrieving a user's permissions and profile metadata from cache, falling back to a database transaction on a miss, and caching it. We also show how to invalidate the cache key when updating user permissions: from aquilia.controller import Controller, GET, PATCH, RequestCtx from aquilia.flow import requires from aquilia.effects import CacheEffect, DBTx class SecurityController(Controller): # Require both cache and db transaction capabilities effects = [ CacheEffect(namespace="permissions"), DBTx["write"] # Write required for update, used as read for retrieve ] @GET("/users/:id/permissions") async def get_permissions(self, id: int, ctx: RequestCtx) -> dict: cache = ctx.get_effect("Cache") db = ctx.get_effect("DBTx") # 1. Attempt to load from the namespaced permissions cache cache_key = f"user: :roles" roles = await cache.get(cache_key) if roles is None: # 2. Cache Miss - Fetch from Database roles = await db.fetch_all( "SELECT role_name FROM user_roles WHERE user_id = ?", (id,) ) # Normalize to basic dictionary list for serialization roles = [dict(r) for r in roles] # 3. Cache results for 1 hour await cache.set(cache_key, roles, ttl=3600) return ctx.json( ) @PATCH("/users/:id/permissions") async def update_permissions(self, id: int, ctx: RequestCtx) -> dict: body = await ctx.json() cache = ctx.get_effect("Cache") db = ctx.get_effect("DBTx") # 1. Update roles in DB await db.execute("DELETE FROM user_roles WHERE user_id = ?", (id,)) await db.execute_many( "INSERT INTO user_roles (user_id, role_name) VALUES (?, ?)", [(id, role) for role in body["roles"]] ) # 2. Invalidate cache key to ensure subsequent requests load fresh state cache_key = f"user: :roles" await cache.delete(cache_key) return ctx.json( ) DBTx Effect Queue Effect )
Go to Homepage