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 native aquilia.http async clients, error handling structures, and status code maps: from aquilia.http import AsyncHTTPClient 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 = AsyncHTTPClient( 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 = await 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) err_text = await response.text() return ProviderResult( status=ProviderResultStatus.PERMANENT_FAILURE, error_message=err_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: err_text = await response.text() 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.close() 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" ) ] )) Mail Authentication & OAuth2 Configure provider authentication via MailAuth helpers: plain, oauth2 (XOAUTH2 for Gmail / Office365), api_key, or ntlm. from aquilia.mail import MailAuth # Standard SMTP authentication auth_plain = MailAuth.plain("username", password_env="SMTP_PASSWORD") # OAuth2 (XOAUTH2) authentication for Gmail or Microsoft 365 auth_oauth = MailAuth.oauth2( user="sender@example.com", access_token_env="GMAIL_OAUTH_TOKEN", ) Shared MIME Construction & DKIM Signing Every provider shares a single MIME builder (aquilia.mail.mime). It handles multipart message formatting, inline attachments, tracking headers (X-Aquilia-Envelope-ID), and DKIM signing. DKIM signing is executed at the byte level before transmission using dkimpy. Enable DKIM in security config; missing keys or dependencies fail the send early so unauthenticated mail is never shipped. MailIntegration( dkim_enabled=True, dkim_domain="example.com", dkim_selector="aquilia", dkim_private_key_env="DKIM_PRIVATE_KEY", ) Install DKIM support via pip install aquilia[mail-dkim]. Validate configuration with aq mail check. 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