Overview — Aquilia Documentation
Comprehensive guide and documentation for Overview in the Aquilia framework. View API reference, examples, and implementation patterns.
interface MiddlewareItem const MIDDLEWARE_DATA: MiddlewareItem[] = [ , , , , , , , , , , , , , ]; const ARCHITECTURE_DETAILS: Record = , manifest: , scanner: , sorting: , compilation: , asgi: , resolver: , ctxpool: , di: , exception: , fault: , scope: , auth: , cache: , controller: , typecheck: , ctxrelease: }; function MiddlewareArchitectureDiagram( : ) ; const details = ARCHITECTURE_DETAILS[selectedNode] || ARCHITECTURE_DETAILS.workspace; // Staggered layout list for Runtime Nodes const runtimeNodes = [ , , , , , , , , , , , ]; return ( System Architecture & Lifecycle Select a phase to explore compiling logic and runtime traversal. 1. Compilation & Sorting 2. Runtime Traversal , , , , ].map((node) => ); })} ) : ( INGRESS GATEWAYS TRAVERSAL PIPELINE (PRIORITIZED CLOSURES) ROUTE CORE ⚡ HEALTH BYPASS (Zero-Alloc Sync Flow) , ].map((sc) => ( ))} else if (node.rail === 'bottom') return ( ); })} )} Target: ); } function MiddlewarePipelineVisualizer( : ) ; }; const getThemeColor = (type: string) => }; const getThemeColorHex = (type: string) => }; return ( Interactive Pipeline Flow Explore details, import paths, and fluent configuration syntax. Spine Flow Onion Rings IMPORT PATH REGISTRATION API )} ); })} ) : ( @keyframes radar-sweep to } .radar-sweep-line CORE else if (mw.type === 'protocol') else if (mw.type === 'security') else if (mw.type === 'core') const nodesOfSameType = MIDDLEWARE_DATA.filter(m => m.type === mw.type); const idx = nodesOfSameType.findIndex(m => m.id === mw.id); const count = nodesOfSameType.length; angle = (idx * (360 / count)) + (mw.type === 'protocol' ? 30 : 0); const = getCoordinates(radius, angle); const isSelected = mw.id === selectedId; const colorHex = getThemeColorHex(mw.type); return ( P ); })} PRIORITY: P DOTTED PATH REGISTRATION )} ); } MIDDLEWARE / OVERVIEW Middleware System & Flows Middleware in Aquilia acts as a series of composable wrappers around the request/response lifecycle. Managed by the MiddlewareStack , every middleware conforms to a strict async signature, enabling deterministic priority execution and scoped filtering. Middleware Pipeline Flow MiddlewareStack .build_handler() sorts every registered descriptor by (scope_rank, priority) ascending, then wraps the final handler in reverse order — so the lowest priority number becomes the outermost layer and runs first on the way in / last on the way out. AquiliaServer._setup_middleware() always registers two internal plumbing middlewares first, regardless of any workspace/module configuration — they are framework infrastructure, not part of the user-facing chain: Note the overlap: if you leave .middleware(...) unset, the server falls back to a hardcoded chain of just ExceptionMiddleware (priority 1) + RequestIdMiddleware (priority 10) — ExceptionMiddleware then sits outside the always-on FaultMiddleware (priority 2) as a last-resort catch-all in case FaultEngine.process() re-raises. See Built-in Middleware for the full, source-verified priority table. Scope Controls Order, Not Which Requests Run It scope ("global" / "app" / "controller" / "route") only feeds MiddlewareStack._sort_middlewares()'s scope_order ranking (global=0, app=1, controller=2, route=3) — it decides where in the wrapping order a middleware sits. build_handler() then wraps every registered descriptor unconditionally; nothing in the stack filters a middleware out for requests outside its declared app or route. MiddlewareConfig.scope_target (intended for "app:name" / "route:/pattern" pinning) is accepted as a field but is never read by MiddlewareStack, MiddlewareConfig.to_dict(), or AquiliaServer._register_app_middleware() — it has no runtime effect. A middleware registered with scope="app" from one module's manifest still runs for every request handled by the process, not just that module's routes. If you need a middleware to run only for a subset of routes, gate it inside __call__ yourself (check request.path / request.state) — don't rely on scope for isolation. ⚠ Auto-Discovery Can Silently Reset Your Manifest Config When AppManifest.auto_discover is True (the default), RuntimeRegistry.perform_autodiscovery() scans your module's package for any class named *Middleware or subclassing Middleware . For every discovered class whose import path lives inside your own module package, it rebuilds a fresh MiddlewareConfig(class_path=...) with default scope="global", priority=50, no config — discarding any custom scope, priority, or constructor config you had explicitly set for that same class in middleware=[MiddlewareConfig(...)]. Middleware pointing at classes outside your module's package (e.g. built-ins from aquilia.middleware_ext) are left untouched — only local, in-package middleware classes are re-discovered and reset. from aquilia.manifest import AppManifest, MiddlewareConfig manifest = AppManifest( name="billing", version="0.1.0", middleware=[ # Custom priority/config here is DISCARDED at startup unless # auto_discover=False, because StripeClientMiddleware lives # inside modules.billing (this module's own package). MiddlewareConfig( class_path="modules.billing.middleware:StripeClientMiddleware", priority=30, config= , ), ], auto_discover=False, # The Middleware Contract Every middleware in Aquilia must inherit from the Middleware base class and implement a callable coroutine signature that accepts exactly three parameters. The signature validation is enforced strictly at startup by the MiddlewareStack : , status=401) return await next_handler(request, ctx) Composing a Production Chain MiddlewareChain ships three presets — .chain() (empty), .minimal(), .defaults() (both identical: ExceptionMiddleware priority 1 + RequestIdMiddleware priority 10), and .production() which additionally adds CompressionMiddleware (priority 15) and TimeoutMiddleware (priority 18, 30s). Start from a preset and append your own entries — the list append order doesn't matter, only priority does: from aquilia.workspace import Workspace from aquilia.integrations import MiddlewareChain workspace = ( Workspace("myapp") .middleware( MiddlewareChain .production() # ExceptionMiddleware(1) + RequestIdMiddleware(10) # + CompressionMiddleware(15) + TimeoutMiddleware(18) .use("modules.auth.middleware:JwtAuthMiddleware", priority=25) .use("aquilia.middleware_ext.RateLimitMiddleware", priority=30, default_limit=200, default_window=60.0) ) ) Remember: FaultMiddleware (priority 2, always on) still wraps everything at priority 1 ExceptionMiddleware in this example, so it fires first when a plain Fault subclass is raised. ExceptionMiddleware is your last-resort net if FaultEngine.process() itself re-raises. Workspace-Level Middleware Config Workspace-level middleware configurations define the global pipeline wrapping all application modules. There are two primary styles to declare these in workspace.py: 1. Fluent MiddlewareChain (Recommended Style) Instantiates the fluent MiddlewareChain directly on the Workspace. This provides auto-completion, priority order sorting, and inline argument checks: from aquilia.workspace import Workspace from aquilia.integrations import MiddlewareChain workspace = ( Workspace("myapp") .middleware( MiddlewareChain() .use("aquilia.middleware.ExceptionMiddleware", priority=1) .use("aquilia.middleware.RequestIdMiddleware", priority=5) .use("modules.auth.middleware:JwtAuthMiddleware", priority=25) ) ) ⚠️ Integration.middleware_chain() + .integrate() — Does Nothing at Runtime Integration.middleware_chain() is a static builder that returns a plain dict tagged _integration_type: "middleware_chain". Passed into Workspace.integrate(), it is stored at self._integrations["middleware_chain"], which serializes into config["integrations"]["middleware_chain"] — not the top-level config["middleware_chain"] key. WARNING (verified from source): Every other integration reader (sessions, cache, storage, ...) uses ConfigLoader.get_subsystem_config(), which falls back from get(name) to get(f"integrations. '}"). But ConfigLoader.get_middleware_config() is hand-written as self.get("middleware_chain") only — it has no fallback to integrations.middleware_chain. Entries configured this way are read by nothing and are silently never instantiated into the running MiddlewareStack. Only Workspace.middleware(MiddlewareChain()) (which sets the dedicated top-level _middleware_chain attribute) actually wires middleware. Do not use Integration.middleware_chain() for anything you need to run. from aquilia.workspace import Workspace from aquilia.integrations import Integration workspace = ( Workspace("myapp") .integrate( # This chain is parsed, stored, and then never read again. Integration.middleware_chain( entries=[ , , ] ) ) ) Module-Level Middleware Config Instead of wrapping the entire workspace, modules declare local middlewares inside their own manifest.py via AppManifest. 1. AppManifest middleware Config (Recommended) Passes list of MiddlewareConfig objects specifying the class path, target scope, priority, and custom parameters: from aquilia.manifest import AppManifest, MiddlewareConfig manifest = AppManifest( name="billing", version="0.1.0", middleware=[ MiddlewareConfig( class_path="modules.billing.middleware:StripeClientMiddleware", scope="app", # Affects wrap ORDER only — see warning above, # it does NOT limit this middleware to "billing" priority=30, config= # Custom constructor kwargs ) ], auto_discover=False, # keep this exact priority/config — see warning above ) ⚠️ AppManifest middlewares list (Deprecated) Legacy manifests declared middleware as a list of tuples containing string paths and configuration dicts. WARNING: AppManifest will issue a deprecation warning at startup when detecting the middlewares attribute and will auto-convert them internally to MiddlewareConfig instances. manifest = AppManifest( name="billing", version="0.1.0", # Legacy tuples configuration middlewares=[ ("modules.billing.middleware:StripeClientMiddleware", ) ] ) MiddlewareStack )
Go to Homepage