Authentication — Aquilia Documentation
Comprehensive guide and documentation for Authentication in the Aquilia framework. View API reference, examples, and implementation patterns.
Security & Authentication AquilAuth — Enterprise-Grade Security Aquilia's authentication architecture is built on three pillars: Identity (the principal), PasswordCredential (the proof of identity), and AuthGuard (policy enforcement). It is async-first, deeply integrated with the dependency injection system, and provides zero-config setups for both API tokens and browser sessions. Authentication Pipeline When a request enters the application, it flows through a unified middleware chain that automatically parses authentication credentials, resolves sessions, and loads the active identity. , , , , , ].map((b, i) => ( } ))} CREDENTIAL TYPES , , , , ].map((c, i) => ( ))} Subsystem Components , , , , , , , , , ].map((m, i) => ( ))} End-to-End Application Integration Rather than configuring independent components, this guide walks you through building a complete user registration, authentication, and endpoint authorization system backed by the database ORM and dependency injection. 01. Define the User DB Model Extend the base Model to define columns for user names, emails, active flags, comma-separated roles, and cryptographically hashed passwords. from aquilia.models import Model from aquilia.models.fields import CharField, EmailField, BooleanField, DateTimeField class User(Model): table = "users" name = CharField(max_length=150) email = EmailField(unique=True) password_hash = CharField(max_length=255) is_active = BooleanField(default=True) roles = CharField(max_length=255, default="user") # e.g. "user,admin" created_at = DateTimeField(auto_now_add=True) 02. Establish Request Validation Contracts Create incoming schema definitions (Contracts) to enforce type validation, boundary checks, and sanitizer casts on incoming API requests automatically. from aquilia.contracts import Contract, Field class SignUpContract(Contract): name = Field[str]() email = Field[str]() password = Field[str]() class SignInContract(Contract): email = Field[str]() password = Field[str]() 03. Implement the Auth Controller Build controller methods to handle user creation, token issuance, and protected dashboard paths. Decorating methods with @authenticated or @require_identity isolates access boundaries. from aquilia.controller import Controller, GET, POST from aquilia.response import Response from aquilia.auth.decorators import authenticated, require_identity from aquilia.auth.manager import AuthManager from aquilia.auth.core import Identity, IdentityType, IdentityStatus from contracts.auth import SignUpContract, SignInContract from models.user import User class AuthController(Controller): prefix = "/auth" @POST("/signup") async def signup(self, ctx, contract: SignUpContract): """Register a new user, hashing their password hash.""" data = contract.validated_data # Verify if record exists existing = await User.objects.filter(email=data["email"]).first() if existing: return Response.json( , status=400) auth_manager = ctx.container.resolve(AuthManager) hashed = auth_manager.password_hasher.hash(data["password"]) user = await User.objects.create( name=data["name"], email=data["email"], password_hash=hashed, is_active=True, roles="user" ) return Response.json( ) @POST("/login") async def login(self, ctx, contract: SignInContract): """Authenticate user credentials and issue security tokens.""" data = contract.validated_data auth_manager = ctx.container.resolve(AuthManager) user = await User.objects.filter(email=data["email"]).first() if not user or not user.is_active: return Response.json( , status=401) # Initialize seed identity for dynamic stores identity = Identity( id=str(user.id), type=IdentityType.USER, attributes= , status=IdentityStatus.ACTIVE, ) try: # Complete login and return Issued Access and Refresh Tokens auth_result = await auth_manager.sign_in( username=user.email, password=data["password"], identity=identity, ) return Response.json( ) except Exception as e: return Response.json( , status=401) @GET("/profile") @authenticated async def profile(self, ctx, user: Identity): """Authenticated endpoint returns current user info.""" return Response.json( ) @GET("/admin-panel") @require_identity(roles=["admin"]) async def admin_panel(self, ctx, user: Identity): """Admin restricted route.""" return Response.json( ) 04. Implement Database Adapters for Auth Storage Map the authentication protocols (`IdentityStore` and `CredentialStore`) directly to your database queries to resolve principals and store password updates securely. from typing import Any from aquilia.auth.core import IdentityStore, CredentialStore, Identity, PasswordCredential, IdentityStatus, IdentityType from models.user import User class DatabaseIdentityStore(IdentityStore): async def get(self, identity_id: str) -> Identity | None: user = await User.objects.filter(id=identity_id).first() if not user or not user.is_active: return None return Identity( id=str(user.id), type=IdentityType.USER, attributes= , status=IdentityStatus.ACTIVE, ) async def get_by_attribute(self, attribute: str, value: Any) -> Identity | None: if attribute == "email": user = await User.objects.filter(email=value).first() if user: return await self.get(str(user.id)) return None class DatabaseCredentialStore(CredentialStore): async def get_password(self, identity_id: str) -> PasswordCredential | None: user = await User.objects.filter(id=identity_id).first() if not user: return None return PasswordCredential( identity_id=identity_id, password_hash=user.password_hash ) async def create_password(self, credential: PasswordCredential) -> None: await User.objects.filter(id=credential.identity_id).update( password_hash=credential.password_hash ) async def update_password(self, credential: PasswordCredential) -> None: await self.create_password(credential) 05. Bootstrap DI Registration and Middleware Register your database store overrides in the dependency injection container, append the AquilAuthMiddleware stack, and launch the application. from aquilia import Aquilia from aquilia.sessions import SessionEngine from aquilia.auth.core import IdentityStore, CredentialStore from aquilia.auth.manager import AuthManager from aquilia.auth.integration.di_providers import register_auth_providers from aquilia.auth.integration.middleware import AquilAuthMiddleware, EnhancedRequestScopeMiddleware from stores.database import DatabaseIdentityStore, DatabaseCredentialStore from controllers.auth import AuthController app = Aquilia() # 1. Register custom DB auth store adapters before initializing auth providers app.container.register(IdentityStore, DatabaseIdentityStore) app.container.register(CredentialStore, DatabaseCredentialStore) # 2. Register all default providers (PasswordHasher, TokenManager, etc.) register_auth_providers(app.container) # 3. Resolve required managers from container session_engine = app.container.resolve(SessionEngine) auth_manager = app.container.resolve(AuthManager) # 4. Attach request scope container creation middleware app.use(EnhancedRequestScopeMiddleware(app.container)) # 5. Apply the main auth middleware to intercept requests app.use(AquilAuthMiddleware( session_engine=session_engine, auth_manager=auth_manager, require_auth=False, # Enforce opt-in per route using decorators )) # 6. Bind controller endpoints app.register_controller(AuthController) Identity Object Structure The principal identity is immutable once instantiated, representing a secure entity descriptor inside RequestCtx. Field Type Description ))} )
Go to Homepage