import logging import os from abc import ABC, abstractmethod from typing import Optional from .base import Provider, ProviderConfig class EmailConfig(ProviderConfig): smtp_server: Optional[str] = None smtp_port: Optional[int] = None smtp_username: Optional[str] = None smtp_password: Optional[str] = None from_email: Optional[str] = None use_tls: Optional[bool] = True sendgrid_api_key: Optional[str] = None mailersend_api_key: Optional[str] = None verify_email_template_id: Optional[str] = None reset_password_template_id: Optional[str] = None password_changed_template_id: Optional[str] = None frontend_url: Optional[str] = None sender_name: Optional[str] = None @property def supported_providers(self) -> list[str]: return [ "smtp", "console", "sendgrid", "mailersend", ] # Could add more providers like AWS SES, SendGrid etc. def validate_config(self) -> None: if ( self.provider == "sendgrid" and not self.sendgrid_api_key and not os.getenv("SENDGRID_API_KEY") ): raise ValueError( "SendGrid API key is required when using SendGrid provider" ) if ( self.provider == "mailersend" and not self.mailersend_api_key and not os.getenv("MAILERSEND_API_KEY") ): raise ValueError( "MailerSend API key is required when using MailerSend provider" ) logger = logging.getLogger(__name__) class EmailProvider(Provider, ABC): def __init__(self, config: EmailConfig): if not isinstance(config, EmailConfig): raise ValueError( "EmailProvider must be initialized with an EmailConfig" ) super().__init__(config) self.config: EmailConfig = config @abstractmethod async def send_email( self, to_email: str, subject: str, body: str, html_body: Optional[str] = None, *args, **kwargs, ) -> None: pass @abstractmethod async def send_verification_email( self, to_email: str, verification_code: str, *args, **kwargs ) -> None: pass @abstractmethod async def send_password_reset_email( self, to_email: str, reset_token: str, *args, **kwargs ) -> None: pass @abstractmethod async def send_password_changed_email( self, to_email: str, *args, **kwargs, ) -> None: pass