Decorators — Aquilia Documentation
Comprehensive guide and documentation for Decorators in the Aquilia framework. View API reference, examples, and implementation patterns.
Cache / Decorators Cache Decorators AquilaCache includes function decorators for declarative read caching and invalidation, plus an HTTP response-cache middleware. @cached Caches function results by key. On a cache miss, it executes the target function, optionally validates the result, and stores it in the cache with the specified TTL and tags. @cached( ttl: int = 300, namespace: str = "default", key: str | None = None, key_func: Callable[..., str] | None = None, # (func, args, kwargs) -> key tags: tuple[str, ...] = (), unless: Callable[..., bool] | None = None, # skip caching if True condition: Callable[[Any], bool] | None = None, # cache only if True ) from aquilia.cache import cached @cached(ttl=60, namespace="api") async def get_popular_products(): return await db.fetch_all("SELECT * FROM products ORDER BY views DESC LIMIT 20") @cached( ttl=300, key_func=lambda func, args, kwargs: f"user: ", condition=lambda result: result is not None, ) async def get_user_profile(user_id: int): return await User.objects.get(id=user_id) @cached( ttl=120, namespace="feed", unless=lambda *args, **kwargs: kwargs.get("no_cache", False), ) async def get_feed(user_id: str, *, no_cache: bool = False): return await feed_repo.fetch(user_id) Changed in 1.3.4 — key generation and None results Two decorator behaviors were corrected. Before 1.3.4 the first positional argument was excluded from every generated key, so all calls to a single-argument function collapsed onto one entry and returned another call's value. Keys are now built from the full call signature. @cached(ttl=60, namespace="users") async def fetch(user_id: int): return await fetch(1) # await fetch(2) # before 1.3.4: Results of None are also cached now (previously they were recomputed on every call, forever). Use condition to opt out where recomputation is intended. key_version and embed the namespace exactly once, matching keys built by CacheService directly. Existing entries written under the old layout become unreachable and expire under their own TTL. If you use @cached on plain functions with a distributed backend, flush the affected namespaces after upgrading. @cache_aside A semantic alias for @cached with identical runtime behavior. Use it to indicate that the decorated function is the authoritative source of truth for the cached data. from aquilia.cache import cache_aside @cache_aside(ttl=180, namespace="products", tags=("products",)) async def find_product(product_id: int): return await Product.objects.get(id=product_id) @invalidate Executes the wrapped function first (typically a write operation), and then invalidates specified keys and/or tags. @invalidate( *keys: str, namespace: str = "default", tags: tuple[str, ...] = (), ) from aquilia.cache import invalidate @invalidate("products:list:v1", namespace="catalog", tags=("products",)) async def create_product(data: dict): return await product_repo.create(data) @invalidate(tags=("products", "catalog:list"), namespace="catalog") async def import_products(batch: list[dict]): return await product_repo.bulk_insert(batch) CacheMiddleware HTTP response cache middleware. Intercepts incoming requests, generates and validates ETags, vary headers, and serves cached response payloads for GET/HEAD methods. CacheMiddleware( cache_service, default_ttl: int = 60, cacheable_methods: tuple[str, ...] = ("GET", "HEAD"), vary_headers: tuple[str, ...] = ("Accept", "Accept-Encoding"), namespace: str = "http_response", stale_while_revalidate: int = 0, cache_authenticated: bool = False, # 1.3.4+ ) Security — identity-aware caching (changed in 1.3.4) The response cache is shared. Before 1.3.4 the default cache key varied only on Accept and Accept-Encoding, so on a per-user route the first authenticated visitor's response was cached and served to everyone else hitting that path until the TTL expired. Two safeguards now apply, and neither can be disabled implicitly: A request carrying Cookie or Authorization bypasses the cache unless that header is listed in vary_headers and cache_authenticated=True is passed. A response that sets Set-Cookie is never stored. Both paths mark the response X-Cache: PRIVATE. Anonymous traffic caches exactly as before. Expect a hit-rate drop on authenticated routes after upgrading — that drop is the leak closing. from aquilia.cache.middleware import CacheMiddleware # Anonymous / public routes -- safe default, nothing extra required server.middleware_stack.add( CacheMiddleware( cache_service=cache_service, default_ttl=60, cacheable_methods=("GET", "HEAD"), vary_headers=("Accept", "Accept-Encoding"), namespace="http", stale_while_revalidate=30, ), scope="global", priority=26, name="cache", ) # Deliberate per-identity caching -- opt in AND vary on the identity header CacheMiddleware( cache_service=cache_service, default_ttl=30, vary_headers=("Accept", "Cookie"), cache_authenticated=True, ) Decorator Cache Resolution Decorators automatically resolve the active CacheService in the following order: Checks for a self.cache attribute on the first argument (typical for controllers). Checks for a self._cache attribute on the first argument. Falls back to the module-level default cache service registered via set_default_cache_service(...). from aquilia.cache.decorators import set_default_cache_service # Optional manual setup if using decorators on standalone helper functions set_default_cache_service(cache_service) )
Go to Homepage