Flow Pipelines — Aquilia Documentation
Comprehensive guide and documentation for Flow Pipelines in the Aquilia framework. View API reference, examples, and implementation patterns.
SUBSYSTEM / FLOW PIPELINES Flow Pipelines The FlowPipeline class compiles and executes a sequence of execution nodes under strict priority bands. It acts as the backbone of request execution and route-specific middleware, orchestrating capability acquisition and safe resource release. Pipeline Anatomy & Execution A pipeline structures your request processing into discrete, composable phases: 1. Guards: Run first to perform security checks, access control, and rate limiting. If a guard returns False or a Response, execution short-circuits. 2. Transforms: Modify request parameters, deserialize payloads, and thread state changes into the context. 3. Effect Acquisition: Automatically lease resource handles from the registry for any capabilities required by subsequent nodes. 4. Handler: Executes core business logic. A pipeline can have exactly one primary handler. 5. Hooks (Post-Handler): Execute post-processing, logging, metrics collection, and can optionally modify the handler's return value. Priority Bands When nodes are added to a pipeline, they are sorted first by their node type band, and then by their individual numeric priority. Aquilia defines standard priority bands as constant integers: PRIORITY_CRITICAL = 10 # Security checks, CORS validation, global rate limiting. PRIORITY_AUTH = 20 # Authentication guards, session lookups, permission validations. PRIORITY_VALIDATE = 30 # Input validation schema checks (Contracts). PRIORITY_TRANSFORM = 40 # Payload transformations, parameter binding. PRIORITY_DEFAULT = 50 # Primary request handler execution. PRIORITY_ENRICH = 60 # Response enrichment, wrapping structures. PRIORITY_LOG = 70 # Audit log generation, performance metrics emission. PRIORITY_CLEANUP = 80 # Post-request teardowns, resource recycling. API Builder Reference def pipeline(name: str = "pipeline", *, timeout: float | None = None) -> FlowPipeline Helper function that instantiates a new FlowPipeline builder. .guard(node, *, name: str | None = None, priority: int = 20, effects: list[str] | None = None, condition: Callable | None = None) Adds a guard node to the pipeline. Guards return True to proceed, or False/a Response to short-circuit. .transform(node, *, name: str | None = None, priority: int = 40, effects: list[str] | None = None) Adds a transformation node that runs after guards. Modifies or enriches context variables. .handler(node, *, name: str | None = None, priority: int = 50, effects: list[str] | None = None) Sets the main execution handler. Receives the context and yields the core response value. .hook(node, *, name: str | None = None, priority: int = 70, effects: list[str] | None = None) Registers post-execution hooks to log telemetry or adjust the resolved response. .compose(*other: FlowPipeline) -> FlowPipeline Merges nodes from multiple pipelines, returning a new pipeline with all nodes sorted by priority. Can also be invoked using the | operator. from_pipeline_list(nodes: Sequence[Any], *, name: str) -> FlowPipeline Utility converting controller pipeline lists into a unified FlowPipeline . Automatically materializes zero-argument factory functions. Building & Executing Pipelines Below is a detailed guide showing how to create a pipeline manually, compose it with operators, and trigger execution within a request context. from aquilia.flow import pipeline, FlowContext, requires from aquilia.response import Response # 1. Define nodes async def check_api_key(ctx: FlowContext): api_key = ctx.request.headers.get("X-API-Key") if api_key != "secret-token": # Returning a Response short-circuits execution return Response.json( , status=401) return True async def sanitize_payload(ctx: FlowContext): # Transforms modify state dictionaries in context if "email" in ctx.state: ctx.state["email"] = ctx.state["email"].strip().lower() @requires("DBTx") async def save_record(ctx: FlowContext): # Automatically acquired DB connection db = ctx.get_effect("DBTx") await db.execute( "INSERT INTO leads (email) VALUES (?)", (ctx.state["email"],) ) return # 2. Build Pipeline lead_pipeline = ( pipeline("create_lead") .guard(check_api_key, priority=10) .transform(sanitize_payload) .handler(save_record) ) # 3. Execute Pipeline # registry contains registered EffectProviders (e.g. DBTxProvider) ctx = FlowContext(request=request, state= ) result = await lead_pipeline.execute(ctx, effect_registry=registry) if result.is_success: print(f"Success! Response: ") elif result.is_guarded: print(f"Guarded (Short-circuited): ") Controller Integrations & Composition You can compose pipelines using the bitwise OR (|) operator. This allows you to define reusable segments (e.g., auth checks) and merge them with route-specific handlers. from aquilia.controller import Controller, POST, RequestCtx from aquilia.flow import pipeline, requires # Reusable security segment auth_pipeline = pipeline("auth_guard").guard(verify_jwt, priority=10) class UserController(Controller): # Controller routes accept pipeline lists @POST( "/users", pipeline=[ auth_pipeline, # Reusable pipeline validate_user_schema, # Plain function (wraps as guard) ] ) @requires("DBTx") async def create_user(self, ctx: RequestCtx) -> dict: # Executes within the compiled pipeline context db = ctx.get_effect("DBTx") ... return Overview Flow Context & Nodes )
Go to Homepage