Subsystem — Aquilia Documentation
Comprehensive guide and documentation for Subsystem in the Aquilia framework. View API reference, examples, and implementation patterns.
SUBSYSTEM / OVERVIEW Subsystem Overview The Subsystem is a unified, strongly-typed architecture combining composable execution flows and declarative capability effects. It compiles request-handling steps into pipelines while lazily resolving infrastructure requirements (like databases or caches) on-demand through an extensible registry. Philosophy & Decoupling In traditional backend frameworks, handlers (e.g., controllers or services) instantiate database connections, cache clients, or message brokers directly. This tight coupling creates mock-heavy tests, configuration drift, and makes it challenging to swap backends (like switching from a local filesystem storage to AWS S3). Aquilia solves this via the Explicit Capability Injection pattern. Handlers declare *what* capability they require via a typed token, and the runtime handles the lifecycle (setup, connection pooling, and automatic commit/rollback cleanup) around the request execution. Request Lifecycle & Flow Architecture The diagram below illustrates how an incoming request flows through the compiled Flow pipeline phases, resolving capability requirements lazily through the registry proxy, and releasing resources safely. HTTP REQUEST Client Inbound FLOWCONTEXT Scoped container GUARDS Auth & validation TRANSFORMS Payload molding EFFECT LEASE Lazy acquisition PIPELINE HANDLER Core execution DISPOSAL & HOOKS Commit / Rollback LIFO Workspace Integration To use the capability system, register EffectMiddleware and FlowContextMiddleware in your application's middleware stack inside your project's workspace configuration (usually in workspace.py). IMPORTANT NOTE ON ORDERING: The FlowContextMiddleware must precede EffectMiddleware in execution (meaning it has a lower priority number or is added first in the chain). This ensures that a request-scoped FlowContext exists so that the Effect Middleware can inject the acquired capability handles into it. from aquilia.workspace import Workspace from aquilia.middleware import MiddlewareChain app = ( Workspace.new("aquilia-project") .middleware( MiddlewareChain.chain() .defaults() .use("aquilia.middleware_ext.FlowContextMiddleware", priority=14) .use("aquilia.middleware_ext.EffectMiddleware", priority=15) ) ) The framework will load the middleware string paths and dynamically inject dependencies. The EffectMiddleware will look for a configured EffectRegistry in the global dependency injection container and defer to it. Declaring Capabilities with @requires To declare that a route handler or flow pipeline node requires a capability, decorate it with the @requires decorator, passing in the names of the required effects (e.g. "DBTx", "Cache"). CRITICAL DECORATOR ORDERING: The @requires(...) decorator must be placed **below** the routing decorators (e.g. @POST or @Get). Python evaluates decorators from bottom to top; applying @requires closest to the method body registers its metadata under __flow_effects__ on the original method, allowing routing compiles to read it. from aquilia.controller import Controller, POST, RequestCtx from aquilia.flow import requires class CorporateOrderIngestController(Controller): @POST("/orders/ingest") @requires("DBTx", "Cache") async def ingest_corporate_order(self, ctx: RequestCtx) -> dict: # Resolve capabilities safely db = ctx.get_effect("DBTx") # Returns a DBTxHandle cache = ctx.get_effect("Cache") # Returns a CacheServiceHandle body = await ctx.json() # 1. Store order payload in Database order_id = await db.fetch_val( "INSERT INTO orders (corporate_id, total, status) VALUES (?, ?, ?) RETURNING id", (body["corporate_id"], body["total"], "RECEIVED") ) # 2. Update cache with hot corporate statistics latest_orders = await cache.get(f"corp: :recent") or [] latest_orders.insert(0, ) await cache.set(f"corp: :recent", latest_orders[:5], ttl=600) return Under The Hood: Deferred Resolving Aquilia uses a *lazy deferred registry* model via the internal proxy _DeferredEffectRegistry. This resolves a fundamental bootstrap ordering constraint: 1. Bootstrap: Middleware stack is constructed early before providers are registered. 2. ASGI Startup: Subsystems load and register their effect providers with the central registry. 3. Request Time: The lazy proxy forwards lookups to the populated live registry on every request. When a request comes in, the Flow pipeline resolves capability requirements dynamically, executes guards and transforms, matches scope permissions, and disposes of transactions on request completion. Flow Pipeline Architecture & Compositions The Flow system enables you to chain independent callable logic blocks (guards, transforms, handlers, and hooks) into a single execution path. By leveraging the bitwise OR (|) operator, you can compose pipelines dynamically, merging reusable middleware blocks with handler nodes. Each phase receives a structured FlowContext containing the request, DI scope, and acquired capability effects: from aquilia.flow import pipeline, FlowContext, requires from aquilia.response import Response # 1. Custom Security Guard Node async def check_api_clearance(ctx: FlowContext) -> bool | Response: clearance = ctx.request.headers.get("X-Clearance-Level") if not clearance or int(clearance) FlowContext: body = await ctx.request.json() ctx.state["raw_payload"] = body ctx.state["processed_at"] = time.time() return ctx # 3. Main Request Handler (Requires Database Capability) @requires("DBTx") async def persist_audit_log(ctx: FlowContext) -> dict: db = ctx.get_effect("DBTx") payload = ctx.state["raw_payload"] log_id = await db.fetch_val( "INSERT INTO audit_logs (level, data) VALUES (?, ?) RETURNING id", (ctx.state["clearance_level"], str(payload)) ) return # 4. Compose pipelines using the '|' operator security_pipeline = pipeline("security").guard(check_api_clearance, priority=10) enrich_pipeline = pipeline("enrich").transform(enrich_payload) full_pipeline = security_pipeline | enrich_pipeline | pipeline("action").handler(persist_audit_log) # 5. Execution and outcome inspection result = await full_pipeline.execute(flow_context, effect_registry=registry) if result.is_success: print(f"Executed handler output: ") elif result.is_guarded: print(f"Short-circuited by guard: ") The EffectProvider Contract Every effect resource must have a corresponding EffectProvider implementation. It defines five lifecycle hooks: async def initialize(self) -> None Invoked once at server startup. Ideal for connecting client pools or setting up driver connections. async def acquire(self, mode: str | None = None) -> Any Invoked per-request. Retrieves the scoped handle representing the resource connection (e.g. transaction, bucket client). async def release(self, resource: Any, success: bool = True) -> None Invoked when a request finishes. The success boolean flags whether the endpoint completed without throwing an exception, allowing providers to safely commit or abort changes. async def finalize(self) -> None Invoked at server shutdown. Closes client connections and drains active pools safely. async def health_check(self) -> dict[str, Any] Aggregates connection health metrics (e.g., checks connection ping, error counts). Returns a dictionary with a "healthy" boolean key. Topological Dependency Resolution (Layer) Aquilia integrates an Effect-TS inspired Layer class that facilitates modular initialization. Layers specify setup factories and declare explicit dependencies. The runtime resolves the full dependency graph topologically at startup. For a detailed guide on managing initialization layers, composing them, and acquiring resources outside of HTTP paths, see the dedicated Layers & Compositions reference. from aquilia.flow import Layer from aquilia.effects import DBTxProvider, CacheProvider # Define configuration dependencies config_layer = Layer( name="Config", factory=lambda: AppConfig.load_from_env() ) db_layer = Layer( name="DBTx", factory=lambda cfg: DBTxProvider(cfg.database_url), deps=["Config"] ) cache_layer = Layer( name="Cache", factory=lambda cfg: CacheProvider(cfg.cache_backend), deps=["Config"] ) # Compose and bootstrap layers sequentially app_layer = Layer.merge(db_layer, cache_layer) Fingerprinting Flow Pipelines )
Go to Homepage