Guards & Context — Aquilia Documentation
Comprehensive guide and documentation for Guards & Context in the Aquilia framework. View API reference, examples, and implementation patterns.
Sessions / Guards Session Guards & Context Secure endpoints declaratively using SessionGuard , compose validations with @requires , and manage scoped sessions inside code blocks using SessionContext context managers. SessionGuard Base Class A SessionGuard handles endpoint authorization checks. Critical: The overridden check() method must return True on success. If it returns None (or does not return), the decorator treats it as falsy and raises an AuthorizationFault. from aquilia.sessions import Session, SessionGuard from aquilia.sessions.faults import SessionPolicyViolationFault class PremiumUserGuard(SessionGuard): """Guard that blocks non-premium users.""" async def check(self, session: Session) -> bool: if not session.get("is_premium"): raise SessionPolicyViolationFault( "Premium subscription required", policy_name="premium_check", ) return True # Critical: Must return True on success! Built-in Auth Guards Session-aware authentication and authorization guards (such as AdminGuard and VerifiedEmailGuard) are defined in the authentication module: AdminGuard from aquilia.auth import AdminGuard, requires from aquilia.sessions import Session from aquilia import Controller, Get class AdminDashboard(Controller): prefix = "/admin" @Get("/") @requires(AdminGuard()) async def dashboard(self, ctx, session: Session): return ctx.json( ) @Get("/users") @requires(AdminGuard()) async def list_users(self, ctx, session: Session): users = await User.objects.all() return ctx.json([u.to_dict() for u in users]) VerifiedEmailGuard from aquilia.auth import VerifiedEmailGuard, requires from aquilia import Controller, Post class SecureController(Controller): prefix = "/secure" @Post("/transfer") @requires(VerifiedEmailGuard()) async def transfer(self, ctx, session): body = await ctx.request.json() return ctx.json( ) @requires Decorator Composition The @requires decorator chains multiple guards on a controller handler. All guards must pass (return True) for the endpoint to execute: from aquilia.auth import requires, AdminGuard, VerifiedEmailGuard from aquilia import Controller, Get, Post class SensitiveController(Controller): prefix = "/sensitive" # Single guard check @Get("/admin-only") @requires(AdminGuard()) async def admin_only(self, ctx, session): return ctx.json( ) # Multiple guards checked in sequence — ALL must pass @Post("/critical-action") @requires(AdminGuard(), VerifiedEmailGuard()) async def critical(self, ctx, session): return ctx.json( ) Custom Session Guards Define custom session-level guards by extending SessionGuard: from aquilia.sessions import SessionGuard from aquilia.sessions.faults import SessionPolicyViolationFault class RoleGuard(SessionGuard): """Enforces specific user roles.""" def __init__(self, role: str): self.role = role async def check(self, session): principal = session.principal if not principal: raise SessionPolicyViolationFault("Authentication required", "role_check") roles = principal.attributes.get("roles", []) if self.role not in roles: raise SessionPolicyViolationFault(f"Role ' ' required", "role_check") return True # Required success return SessionContext The SessionContext manager provides scoped asynchronous context managers. They accept the request context (ctx) and handle startup resolution and shutdown commit/rollbacks: from aquilia.sessions import SessionContext # 1. .authenticated() — Context block requiring authentication async def protected_operation(ctx): async with SessionContext.authenticated(ctx) as session: user_id = session.principal.id session["last_action"] = "protected_op" # Committed automatically on exiting the context block successfully # 2. .ensure() — Context block ensuring a session exists async def track_visitor(ctx): async with SessionContext.ensure(ctx) as session: session["visits"] = session.get("visits", 0) + 1 # 3. .transactional() — Context block with automatic snapshot-rollback on exceptions async def critical_update(ctx): async with SessionContext.transactional(ctx) as session: session["balance"] -= 100 # If any exception is raised here, session data is restored to its original snapshot state await process_external_billing() )
Go to Homepage