API Reference¶

def create(
    self,
    subject: str,
    recipients: List[str],
    body: str,
    sender: Optional[str] = None,
    cc: Optional[List[str]] = None,
    bcc: Optional[List[str]] = None,
    reply_to: Optional[str] = None,
    attachments: Optional[List[str]] = None,
    html_body: bool = False,
    plain_body: Optional[str] = None,
    in_reply_to: Optional[str] = None,
    references: Optional[List[str]] = None,
    priority: Optional[str] = None,
    unsubscribe_url: Optional[str] = None,
    inline_images: Optional[Dict[str, str]] = None,
    inline_css: bool = False,
) -> "FluxMail":
    """Creates an email with the specified details.

    Parameters
    ----------
    subject : str
        Email subject.
    recipients : list of str
        Recipient email addresses.
    body : str
        Email content (plain text or HTML).
    sender : str, optional
        Sender address (SMTP only). Defaults to ``username`` if ``username``
        is a valid email address, otherwise this parameter is required.
    cc : list of str, optional
        CC'd email addresses.
    bcc : list of str, optional
        BCC'd email addresses (SMTP only).
    reply_to : str, optional
        Reply-To address (SMTP only).
    attachments : list of str, optional
        Paths to files to attach.
    html_body : bool, optional
        Use HTML format. Default: ``False``.
    plain_body : str, optional
        Plain-text fallback body. When provided alongside ``html_body=True``
        the message is built as ``multipart/alternative``.
    in_reply_to : str, optional
        Message-ID of the email being replied to (SMTP only).
    references : list of str, optional
        List of Message-IDs forming the thread chain (SMTP only).
    priority : str, optional
        Message priority: ``"high"``, ``"normal"``, or ``"low"`` (SMTP only).

    Returns
    -------
    FluxMail
        Returns ``self`` to support method chaining.
    """
    # Reset so create() can be called again on the same instance.
    if self.is_smtp():
        self.message = EmailMessage()  # type: ignore[assignment]
    self.is_created = False

    self.subject = subject
    self.recipients = recipients
    self.body = body  # type: ignore[assignment]
    self.plain_body = plain_body  # type: ignore[assignment]
    self.sender = sender  # type: ignore[assignment]
    self.cc = cc  # type: ignore[assignment]
    self.bcc = bcc  # type: ignore[assignment]
    self.reply_to = reply_to  # type: ignore[assignment]
    self.input_path = attachments
    self.html_body = html_body  # type: ignore[assignment]
    self.in_reply_to = in_reply_to  # type: ignore[assignment]
    self.references = references  # type: ignore[assignment]
    self.priority = priority  # type: ignore[assignment]
    self.unsubscribe_url = unsubscribe_url  # type: ignore[assignment]
    self.inline_images = inline_images
    self.inline_css = inline_css

    self._validate_parameters()
    self._handle_message_id()
    self._handle_sender()
    self._handle_recipient()
    self._handle_cc()
    self._handle_bcc()
    self._handle_reply_to()
    self._handle_unsubscribe()
    self._handle_threading()
    self._handle_priority()
    self._set_content_type()
    self._attach_files()
    self._attach_inline_images()
    self.is_created = True
    self.logger.info(
        "Email created: subject=%r to=%d cc=%d bcc=%d attachments=%d html=%s",
        self.subject,
        len(self.recipients),
        len(self.cc) if self.cc else 0,
        len(self.bcc) if self.bcc else 0,
        len(self.input_path) if self.input_path else 0,
        self.html_body,
    )
    return self

def display(self) -> str:  # type: ignore[return]
    """Displays or returns an email preview.

    Returns
    -------
    str
        Email preview string.

    Raises
    ------
    FluxMailException
        If ``create()`` has not been called or display fails.
    """
    if not self.is_created:
        raise FluxMailException(
            "Call create() before display().", code="not_created"
        )
    try:
        if self.is_smtp():
            return f"Email Preview:\n{self.message}"
        elif self.is_outlook():
            self.message.Display()  # type: ignore[union-attr]
            return "Outlook email displayed successfully."
    except FluxMailException:
        raise
    except Exception as e:
        msg = f"Display failed: {e}"
        self.logger.error(msg)
        raise FluxMailException(msg, code="display_failed") from e

def send(self, dry_run: bool = False) -> str:  # type: ignore[return]
    """Sends or previews the email.

    Parameters
    ----------
    dry_run : bool, optional
        If ``True``, display the email instead of sending it. Default: ``False``.

    Returns
    -------
    str
        Success or preview message.

    Raises
    ------
    FluxMailException
        If ``create()`` has not been called, or on send failure.
    """
    if not self.is_created:
        raise FluxMailException("Call create() before send().", code="not_created")
    if dry_run:
        return self.display()  # type: ignore[union-attr]
    if self.is_smtp() and not self.host.relay:
        raise FluxMailException("No SMTP relay configured.", code="no_relay")

    self.logger.info(
        "Sending email: subject=%r to=%r relay=%s",
        self.subject,
        self.recipients,
        self.host.relay,
    )
    try:
        if self.is_smtp():
            if self.max_retries > 0:
                for attempt in Retrying(
                    stop=stop_after_attempt(self.max_retries + 1),
                    wait=wait_fixed(self.retry_delay),
                    reraise=True,
                    before_sleep=lambda rs: self.logger.warning(
                        "Send attempt %d failed (%s) — retrying in %.1fs",
                        rs.attempt_number,
                        rs.outcome.exception(),  # type: ignore[union-attr]
                        self.retry_delay,
                    ),
                ):
                    with attempt:
                        self._transport.send(self.message)
            else:
                self._transport.send(self.message)
            self.logger.info(
                "Email sent successfully: subject=%r to=%r",
                self.subject,
                self.recipients,
            )
            return "Email sent successfully via SMTP."
        elif self.is_outlook():
            raise FluxMailException(
                "Outlook requires user interaction to send emails and cannot "
                "send programmatically.",
                code="outlook_no_send",
            )
    except FluxMailException:
        raise
    except Exception as e:
        msg = f"Send failed: {e}"
        self.logger.error(msg)
        raise FluxMailException(msg, code="send_failed") from e

async def send_async(self, dry_run: bool = False) -> str:  # type: ignore[return]
    """Sends or previews the email asynchronously (SMTP only).

    Note: retry (max_retries) is not supported for async sends.
    """
    if not self.is_created:
        raise FluxMailException(
            "Call create() before send_async().", code="not_created"
        )
    if dry_run:
        return self.display()
    if self.is_smtp() and not self.host.relay:
        raise FluxMailException("No SMTP relay configured.", code="no_relay")

    self.logger.info(
        "Sending email async: subject=%r to=%r relay=%s",
        self.subject,
        self.recipients,
        self.host.relay,
    )
    try:
        if self.is_smtp():
            await self._transport.send_async(self.message)
            self.logger.info(
                "Email sent successfully (async): subject=%r to=%r",
                self.subject,
                self.recipients,
            )
            return "Email sent successfully via SMTP."
        elif self.is_outlook():
            raise FluxMailException(
                "Outlook does not support async sending.",
                code="outlook_no_async",
            )
    except FluxMailException:
        raise
    except Exception as e:
        msg = f"Async send failed: {e}"
        self.logger.error(msg)
        raise FluxMailException(msg, code="send_failed") from e

@classmethod
def from_env(cls) -> "FluxMail":
    """Create a ``FluxMail`` instance from environment variables.

    Reads ``FLUXMAIL_TYPE``, ``FLUXMAIL_HOST``, ``FLUXMAIL_PORT``,
    ``FLUXMAIL_USERNAME``, ``FLUXMAIL_PASSWORD``, ``FLUXMAIL_TLS``,
    ``FLUXMAIL_SSL``, ``FLUXMAIL_TIMEOUT``, ``FLUXMAIL_MAX_RETRIES``,
    ``FLUXMAIL_RETRY_DELAY``.

    Raises
    ------
    FluxMailException
        If ``FLUXMAIL_HOST`` is missing when ``FLUXMAIL_TYPE=smtp``.
    """
    object_type = os.environ.get("FLUXMAIL_TYPE", "smtp")
    host_str = os.environ.get("FLUXMAIL_HOST", "")
    if object_type.lower() == "smtp" and not host_str:
        raise FluxMailException(
            "FLUXMAIL_HOST is required when FLUXMAIL_TYPE=smtp",
            code="invalid_config",
        )
    host = host_str if host_str else EmailInstance(relay="")
    return cls(
        object_type=object_type,
        host=host,
        port=int(os.environ.get("FLUXMAIL_PORT", "25")),
        username=os.environ.get("FLUXMAIL_USERNAME"),
        password=os.environ.get("FLUXMAIL_PASSWORD"),
        use_tls=_parse_bool(os.environ.get("FLUXMAIL_TLS", "false")),
        use_ssl=_parse_bool(os.environ.get("FLUXMAIL_SSL", "false")),
        timeout=int(os.environ.get("FLUXMAIL_TIMEOUT", "30")),
        max_retries=int(os.environ.get("FLUXMAIL_MAX_RETRIES", "0")),
        retry_delay=float(os.environ.get("FLUXMAIL_RETRY_DELAY", "1.0")),
    )

def test_connection(self) -> dict:
    """Test SMTP connectivity and authentication without sending email.

    Returns
    -------
    dict
        ``{"ok": True, "relay": str, "port": int, "latency_ms": int}``

    Raises
    ------
    FluxMailException
        ``code="outlook_no_connect"`` for Outlook instances.
        ``code="connection_failed"`` if connection or auth fails.
    """
    if self.is_outlook():
        raise FluxMailException(
            "Outlook does not support connection testing.",
            code="outlook_no_connect",
        )
    try:
        start = time.monotonic()
        conn = self._transport._make_connection()
        latency_ms = int((time.monotonic() - start) * 1000)
        try:
            conn.quit()
        except Exception:
            pass
        self.logger.info(
            "Connection test OK: relay=%s port=%d latency=%dms",
            self.host.relay,
            self.port,
            latency_ms,
        )
        return {
            "ok": True,
            "relay": self.host.relay,
            "port": self.port,
            "latency_ms": latency_ms,
        }
    except FluxMailException:
        raise
    except Exception as e:
        msg = f"Connection test failed: {e}"
        self.logger.error(msg)
        raise FluxMailException(msg, code="connection_failed") from e