Source code for litestar_email.plugin

from typing import TYPE_CHECKING

from litestar.plugins import InitPluginProtocol

if TYPE_CHECKING:
    from litestar.config.app import AppConfig
    from litestar.datastructures import State

    from litestar_email.backends.base import BaseEmailBackend
    from litestar_email.config import EmailConfig
    from litestar_email.service import EmailService

__all__ = ("EmailPlugin",)


[docs] class EmailPlugin(InitPluginProtocol): """Litestar plugin for email functionality. This plugin provides email sending capabilities for Litestar applications. It supports multiple backend providers (SMTP, SendGrid, Resend, etc.) through a pluggable backend system. Example: Basic plugin configuration:: from litestar import Litestar from litestar_email import EmailConfig, EmailPlugin, SMTPConfig email_config = EmailConfig( backend=SMTPConfig(host="localhost", port=1025), from_email="noreply@example.com", ) app = Litestar(plugins=[EmailPlugin(config=email_config)]) Dependency injection in route handlers:: from litestar import get from litestar_email import EmailMessage, EmailService @get("/welcome/{email:str}") async def send_welcome(email: str, mailer: EmailService) -> dict[str, str]: await mailer.send_message( EmailMessage(subject="Welcome!", body="Thanks for signing up.", to=[email]), ) return {"status": "sent"} Standalone usage with context manager:: from litestar_email import EmailConfig, EmailMessage, SMTPConfig config = EmailConfig( backend=SMTPConfig(host="localhost", port=1025), from_email="noreply@example.com", ) async with config.provide_service() as mailer: await mailer.send_message( EmailMessage(subject="Hello", body="World", to=["user@example.com"]), ) Using with event listeners:: Note: Event listeners in Litestar execute outside request context and cannot receive DI-injected dependencies. Pass the mailer explicitly. from litestar import get from litestar.events import listener from litestar_email import EmailMessage, EmailService @get("/register/{email:str}") async def register(email: str, mailer: EmailService) -> dict[str, str]: # Pass the DI-injected mailer to the event request.app.emit("user.registered", email, mailer=mailer) return {"status": "queued"} @listener("user.registered") async def on_user_registered(email: str, mailer: EmailService) -> None: # mailer is passed explicitly from emit(), not injected via DI await mailer.send_message( EmailMessage(subject="Welcome!", body="Thanks!", to=[email]), ) """ __slots__ = ("_config",)
[docs] def __init__(self, config: "EmailConfig | None" = None) -> None: """Initialize the email plugin. Args: config: Optional email configuration. If not provided, defaults will be used. """ from litestar_email.config import EmailConfig self._config = config or EmailConfig()
@property def config(self) -> "EmailConfig": """Return the plugin configuration. Returns: The email configuration instance. """ return self._config
[docs] def get_service(self, state: "State | None" = None) -> "EmailService": """Return an EmailService for this plugin. Args: state: Optional application state to fetch a cached service instance or config. Returns: An EmailService instance. """ return self._config.get_service(state)
[docs] def get_backend( self, fail_silently: bool | None = None, ) -> "BaseEmailBackend": """Return a backend instance configured for this plugin. Args: fail_silently: Optional override for fail_silently behavior. Returns: A configured backend instance. """ return self._config.get_backend(fail_silently=fail_silently)
[docs] def on_app_init(self, app_config: "AppConfig") -> "AppConfig": """Handle application initialization. This method is called during Litestar application initialization. Registers the EmailService dependency, signature namespace entries, and app state helpers for use in handlers and event listeners. Args: app_config: The Litestar application configuration. Returns: The application configuration (unmodified). """ app_config.dependencies.update(self._config.dependencies) app_config.signature_namespace.update(self._config.signature_namespace) app_config.state.update({self._config.email_service_state_key: self._config}) return app_config