Queue & Task Effects — Aquilia Documentation
Comprehensive guide and documentation for Queue & Task Effects in the Aquilia framework. View API reference, examples, and implementation patterns.
EFFECTS / QUEUE & TASKS Queue & Task Effects The QueueEffect provides messaging capability inside route handlers. It can pub/sub events to standard brokers via the QueueProvider or enqueue async background tasks via the TaskQueueProvider . Message Queue Publishing When wired with the standard QueueProvider , the effect returns a QueueHandle . This handle facilitates publishing messages to brokers like Redis Streams or RabbitMQ. During testing, it falls back to collecting messages in an in-memory list. QueueHandle API await handle.publish(payload: Any, *, headers: dict[str, str] | None = None) -> None Publishes a single event payload to the configured topic, attaching optional metadata headers. await handle.publish_batch(payloads: Sequence[Any]) -> None Publishes a list of payloads sequentially, optimizing connection roundtrips. from aquilia.controller import Controller, POST, RequestCtx from aquilia.flow import requires from aquilia.effects import QueueEffect class TelemetryIngestController(Controller): # Require Queue capability scoped to the "telemetry" topic effects = [QueueEffect("device_metrics")] @POST("/ingest/metrics") async def ingest_device_data(self, ctx: RequestCtx) -> dict: body = await ctx.json() queue = ctx.get_effect("Queue") # QueueHandle instance # Format payload and headers for real-world RabbitMQ/Redis Broker ingestion payload = await queue.publish( payload, headers= ) return Background Task Worker Integration When registered as a TaskQueueProvider , the effect hooks directly into Aquilia's background worker system. Rather than executing logic inside the HTTP request loop, it allows handlers to defer complex, resource-heavy operations (e.g. sending transaction emails or running AI inference models) to worker processes. It yields a request-scoped TaskQueueHandle that exposes the enqueue method. TaskQueueHandle API await handle.enqueue(func: Any, *args: Any, **kwargs: Any) -> str Submits an asynchronous function or task import path to the queue. Returns the unique Job ID. await handle.publish(payload: Any, *, headers: dict | None = None) -> None Compatibility no-op method designed to make the handle interchangeable with QueueHandle. from aquilia.controller import Controller, POST, RequestCtx from aquilia.flow import requires from aquilia.effects import QueueEffect class RegistrationController(Controller): # Triggers task queue handle acquisition effects = [QueueEffect("default")] @POST("/register/corporate") async def register_company(self, ctx: RequestCtx) -> dict: body = await ctx.json() task_queue = ctx.get_effect("Queue") # TaskQueueHandle instance # Dispatch background worker chain (invoice rendering + email delivery) job_id = await task_queue.enqueue( "modules.billing.tasks:process_corporate_enrollment", company_id=body["company_id"], plan_tier=body["plan_tier"], billing_email=body["billing_email"] ) return Cache Effect HTTP Effect )
Go to Homepage