Providers — Aquilia Documentation
Comprehensive guide and documentation for Providers in the Aquilia framework. View API reference, examples, and implementation patterns.
Mail / Providers Mail Providers & Backends All email backends implement the standard IMailProvider protocol, allowing you to swap backends between local files, stdout consoles, and cloud providers (SMTP, SES, SendGrid) without changing your application code. Pluggable Backends Comparison , , , , ].map((item, i) => ( deps: ))} IMailProvider Custom Implementation To write a custom provider (e.g., Postmark API), implement the IMailProvider interface using `httpx` async clients, error handling structures, and status code maps: import httpx from aquilia.mail.providers import IMailProvider, ProviderResult, ProviderResultStatus from aquilia.mail import MailEnvelope class PostmarkMailProvider(IMailProvider): name = "postmark" supports_batching = False max_batch_size = 1 def __init__(self, api_token: str): self.api_token = api_token self.client = None async def initialize(self) -> None: # Open a persistent connection pool with server credentials self.client = httpx.AsyncClient( headers= , timeout=10.0 ) async def send(self, envelope: MailEnvelope) -> ProviderResult: payload = try: response = await self.client.post("https://api.postmarkapp.com/email", json=payload) if response.status_code == 200: data = response.json() return ProviderResult( status=ProviderResultStatus.SUCCESS, provider_message_id=data.get("MessageID") ) elif response.status_code == 422: # Permanent failure (e.g. invalid recipient address format) return ProviderResult( status=ProviderResultStatus.PERMANENT_FAILURE, error_message=response.text ) elif response.status_code == 429: # Rate limited, request retry backoff duration return ProviderResult( status=ProviderResultStatus.RATE_LIMITED, retry_after=float(response.headers.get("Retry-After", 60)) ) else: return ProviderResult( status=ProviderResultStatus.TRANSIENT_FAILURE, error_message=f"HTTP Error : " ) except Exception as e: # Handle socket/timeout exceptions as transient retry attempts return ProviderResult( status=ProviderResultStatus.TRANSIENT_FAILURE, error_message=str(e) ) async def health_check(self) -> bool: try: res = await self.client.get("https://status.postmarkapp.com/api/v2/status.json") return res.status_code == 200 except Exception: return False async def shutdown(self) -> None: if self.client: await self.client.aclose() Provider Configurations Configure individual providers as list parameters inside the typed MailIntegration : from aquilia.integrations import ( MailIntegration, SmtpProvider, SesProvider, SendGridProvider ) workspace.integrate(MailIntegration( providers=[ # 1. Standard SMTP Server SmtpProvider( host="smtp.gmail.com", port=587, use_tls=True, timeout=15 ), # 2. Amazon SES SesProvider( region="us-east-1", aws_access_key_id="AKIA...", aws_secret_access_key="secret" ), # 3. SendGrid SendGridProvider( api_key="SG.xxx" ) ] )) MailProviderRegistry & Auto-Discovery The mail subsystem leverages Aquilia's PackageScanner to auto-discover provider classes. Register custom provider packages to make them auto-wireable: from aquilia.mail.di_providers import MailProviderRegistry registry = MailProviderRegistry() # 1. Register custom package to scan for IMailProvider classes registry.add_scan_package("myapp.mail_providers") # 2. Discover classes and get custom mapping discovered_types = registry.discover() # Result: MailService Mail Templates )
Go to Homepage