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) @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, ) from aquilia.cache.middleware import CacheMiddleware server.middleware_stack.add( CacheMiddleware( cache_service=cache_service, default_ttl=60, cacheable_methods=("GET", "HEAD"), vary_headers=("Accept", "Accept-Encoding", "Authorization"), namespace="http", stale_while_revalidate=30, ), scope="global", priority=26, name="cache", ) 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