Skip to main content

EmailQueueProtocol

Protocol for email queue operations.

This is the PORT in the ports-and-adapters pattern.
Implementations (adapters) handle the actual queuing mechanism.

Adapters:
- DatabaseEmailQueueAdapter: Stores in EmailQueue DocType
- NotificationServiceAdapter: Calls external notification service
- InMemoryEmailQueueAdapter: For testing

Example:
    # Get adapter from DI container
    queue: EmailQueueProtocol = container.email_queue()

    # Queue an email
    result = await queue.queue_email(EmailRequest(
        to=["user@example.com"],
        subject="Welcome!",
        body="<h1>Hello</h1>",
    ))

Source: email_queue.py

Methods​

queue_email​

async def queue_email(self, request: EmailRequest) -> EmailQueueResult

Queue an email for sending.

    Args:
        request: Email request with recipients, subject, body, etc.

    Returns:
        EmailQueueResult with queue_id and status

    Raises:
        EmailQueueError: If queuing fails

get_status​

async def get_status(self, queue_id: str) -> str | None

Get the status of a queued email.

    Args:
        queue_id: The queue ID from EmailQueueResult

    Returns:
        Status string (Queued, Sending, Sent, Failed) or None if not found

cancel​

async def cancel(self, queue_id: str) -> bool

Cancel a queued email (if not yet sent).

    Args:
        queue_id: The queue ID from EmailQueueResult

    Returns:
        True if cancelled, False if already sent or not found