Flow Context & Nodes — Aquilia Documentation
Comprehensive guide and documentation for Flow Context & Nodes in the Aquilia framework. View API reference, examples, and implementation patterns.
SUBSYSTEM / CONTEXT & NODES Flow Context & Nodes Deep-dive into the data carriers and execution units of the Flow system. Learn how FlowContext threads state and resources through FlowNode pipelines, returning detailed execution results. FlowContext Anatomy The FlowContext threads execution state, request parameters, dependency containers, and acquired effect resources through every stage of the pipeline. class FlowContext: request: Any # The raw ASGI/HTTP request context (e.g. RequestCtx). container: Any # The request-scoped dependency injection container. state: dict[str, Any] # Arbitrary mutable key-value storage used by transforms and handlers. identity: Any # The authenticated principal, typically resolved and set by auth guards. session: Any # Active session state proxy if session middleware is active. effects: dict[str, Any] # Acquired capability resource handles (e.g. database transaction, cache client). metadata: dict[str, Any] # Telemetry logs tracking timings, executed node traces, and acquired effects. Context Method Reference def get_effect(self, name: str) -> Any: Retrieves an acquired capability resource. If the resource is not active, it throws an EffectNotAcquiredFault. Supports type overloading for standard effects. def has_effect(self, name: str) -> bool: Returns whether the specified effect capability is currently acquired and bound to the context. def add_cleanup(self, callback: Callable[[], Awaitable[None]]) -> None: Registers an asynchronous teardown callback. Cleanup actions are executed in Last-In-First-Out (LIFO) order during pipeline disposal. def dispose(self) -> None: Drives the execution of all registered cleanup callbacks, ensuring database transactions roll back or temp files clean up on pipeline failure. Flow Nodes & Node Types A FlowNode represents a single callable unit in a pipeline. The pipeline compiler parses callables, extracts decorator metadata, and maps them to concrete nodes: class FlowNodeType(Enum): GUARD = "guard" # Short-circuits execution (e.g. auth / input validation). TRANSFORM = "transform" # Modifies request parameters or updates context state. HANDLER = "handler" # Core business logic method (usually a controller route). HOOK = "hook" # Post-processing hooks running after the main handler. EFFECT = "effect" # Managed capability resources acquired lazily. MIDDLEWARE = "middleware" # Wraps the entire execution chain. Declaring Requirements with @requires The @requires decorator binds capability metadata directly onto the decorated callable function. When a pipeline executes, it crawls the node list, inspects the callables for the __flow_effects__ property, and triggers batch acquisition before the handler runs. from aquilia.flow import requires, FlowContext @requires("DBTx", "Cache") async def process_payment(ctx: FlowContext): # Retrieve capabilities safely db = ctx.get_effect("DBTx") cache = ctx.get_effect("Cache") # execute database queries user_id = ctx.state["user_id"] balance = await db.fetch_val("SELECT balance FROM accounts WHERE user_id = ?", (user_id,)) ... return FlowResult & Execution Outcomes Triggering execute() returns a structured FlowResult instance representing the final state: class FlowStatus(Enum): SUCCESS = "success" # Completed successfully; result.value contains the handler output. GUARDED = "guarded" # Short-circuited by a guard; result.guard contains the guard node. ERROR = "error" # Unhandled exception raised; result.error contains the exception. TIMEOUT = "timeout" # Execution duration exceeded configured timeout limit. CANCELLED = "cancelled" # Pipeline task cancelled before completion. Context Cleanup Callback Flow You can register custom teardown code directly onto the context. This guarantees cleanup runs even if subsequent nodes throw errors or timeout: import os from aquilia.flow import FlowContext, FlowError async def write_temp_file(ctx: FlowContext): temp_path = f"/tmp/process_ .json" # 1. Write initial payload with open(temp_path, "w") as f: f.write(ctx.state["payload"]) # 2. Register callback to delete file on teardown async def cleanup_temp(): if os.path.exists(temp_path): os.remove(temp_path) ctx.add_cleanup(cleanup_temp) ctx.state["temp_file"] = temp_path # Teardown triggers LIFO order # await ctx.dispose() Flow Pipelines Layers & Compositions )
Go to Homepage