Custom Effects — Aquilia Documentation
Comprehensive guide and documentation for Custom Effects in the Aquilia framework. View API reference, examples, and implementation patterns.
EFFECTS / CUSTOM EFFECTS Custom Effects & Providers Aquilia's Effect system is open and fully extensible. By subclassing Effect and EffectProvider , you can declare custom infrastructural integrations (e.g. SMTP clients, payment gateways, or AI models) and inject them safely into request contexts. Step-by-Step Implementation To extend the capability system, you must define three items: Effect Token: A subclass of Effect representing the typed capability tag. Resource Handle: A class wrapping the provider client. Handlers interact with this handle inside the request context. Effect Provider: A subclass of EffectProvider managing the connection, setup, and cleanup lifecycle. Code Implementation (Slack Dispatcher) Here is how to create a custom SlackEffect that pools HTTPS webhook client sessions, formats slack block payloads, and manages channel routing: import aiohttp from typing import Any, dict from aquilia.effects import Effect, EffectProvider, EffectKind # 1. Define the Effect Token class SlackEffect(Effect[str]): def __init__(self, channel: str = "general"): super().__init__("Slack", mode=channel, kind=EffectKind.CUSTOM) # 2. Define the user-facing resource handle class SlackHandle: def __init__(self, session: aiohttp.ClientSession, webhook_url: str, channel: str): self.session = session self.webhook_url = webhook_url self.channel = channel async def send_message(self, text: str, blocks: list[dict] | None = None) -> None: payload = ", "text": text, "blocks": blocks or [] } async with self.session.post(self.webhook_url, json=payload) as resp: resp.raise_for_status() # 3. Define the Lifecycle Provider class SlackProvider(EffectProvider): def __init__(self, webhook_url: str): self.webhook_url = webhook_url self.session = None async def initialize(self) -> None: # Invoked once at server startup: initialize shared async HTTP session self.session = aiohttp.ClientSession() async def acquire(self, mode: str | None = None) -> SlackHandle: # Invoked per-request: return a handle scoped to the requested channel mode channel = mode or "general" return SlackHandle(self.session, self.webhook_url, channel) async def release(self, resource: SlackHandle, success: bool = True) -> None: # Invoked per-request end: nothing to tear down since session is pooled pass async def finalize(self) -> None: # Invoked once at server shutdown: safely close async HTTP session if self.session: await self.session.close() async def health_check(self) -> dict[str, Any]: if not self.session or self.session.closed: return return Provider Registration You can register your custom provider with the EffectRegistry during application startup: from aquilia.effects import EffectRegistry from extensions.slack_effect import SlackProvider # Create or fetch registry registry = EffectRegistry() # Register custom provider with Slack webhook URL configuration registry.register("Slack", SlackProvider(webhook_url="https://hooks.slack.com/services/T00/B00/X00")) Using in Controllers Once registered, require the capability by name. The handle will be automatically acquired and injected into the context: from aquilia.controller import Controller, POST, RequestCtx from aquilia.flow import requires from extensions.slack_effect import SlackEffect class SlackNotificationController(Controller): # Require Slack capability scoped to the "incidents" channel effects = [SlackEffect("incidents")] @POST("/notify/incident") async def dispatch_incident(self, ctx: RequestCtx) -> dict: slack = ctx.get_effect("Slack") # SlackHandle instance # Dispatch formatted alert block await slack.send_message( text="ALERT: Incident detected in server-prod-04", blocks=[ } ] ) return Storage Effect )
Go to Homepage