tessl/pypi-aiosmtplib

Asyncio SMTP client for sending emails asynchronously in Python applications

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

SMTP Client Class

Name: tessl/pypi-aiosmtplib
Author: tessl

The SMTP class provides a full-featured async SMTP client with persistent connection management, comprehensive SMTP command support, and fine-grained control over email operations. It's ideal for applications that need to send multiple emails, require custom SMTP operations, or need detailed control over the connection lifecycle.

Capabilities

Client Initialization and Connection

Create and configure SMTP client instances with comprehensive connection options.

class SMTP:
    def __init__(
        self,
        *,
        hostname: Optional[str] = None,
        port: Optional[int] = None,
        username: Optional[Union[str, bytes]] = None,
        password: Optional[Union[str, bytes]] = None,
        local_hostname: Optional[str] = None,
        source_address: Optional[tuple[str, int]] = None,
        timeout: Optional[float] = 60,
        use_tls: bool = False,
        start_tls: Optional[bool] = None,
        validate_certs: bool = True,
        client_cert: Optional[str] = None,
        client_key: Optional[str] = None,
        tls_context: Optional[ssl.SSLContext] = None,
        cert_bundle: Optional[str] = None,
        socket_path: Optional[SocketPathType] = None,
        sock: Optional[socket.socket] = None,
    ) -> None:
        """
        Initialize SMTP client with connection parameters.
        
        Parameters can be provided at initialization or when calling connect().
        Options provided to connect() override initialization values except timeout.
        """

    async def connect(
        self,
        *,
        hostname: Optional[str] = None,
        port: Optional[int] = None,
        source_address: Optional[tuple[str, int]] = None,
        timeout: Optional[float] = None,
        socket_path: Optional[SocketPathType] = None,
        sock: Optional[socket.socket] = None,
    ) -> SMTPResponse:
        """
        Connect to SMTP server and perform initial handshake.
        
        Returns:
        SMTPResponse from the server greeting
        
        Raises:
        - SMTPConnectError: If connection fails
        - SMTPConnectTimeoutError: If connection times out
        - SMTPConnectResponseError: If server greeting is invalid
        """

    def close(self) -> None:
        """Close the connection immediately without QUIT command."""

    async def quit(self) -> SMTPResponse:
        """Send QUIT command and close connection gracefully."""

Context Manager Support

The SMTP client supports async context manager for automatic connection management.

async def __aenter__(self) -> "SMTP":
    """Enter async context manager, connecting if not already connected."""

async def __aexit__(
    self,
    exc_type: Optional[type[BaseException]],
    exc_val: Optional[BaseException],
    exc_tb: Optional[TracebackType],
) -> None:
    """Exit async context manager, closing connection gracefully."""

Connection Status and Properties

Check connection status and access server capabilities.

@property
def is_connected(self) -> bool:
    """True if connected to SMTP server."""

@property
def last_ehlo_response(self) -> Union[SMTPResponse, None]:
    """Last EHLO response from server, None if EHLO not performed."""

@property
def is_ehlo_or_helo_needed(self) -> bool:
    """True if EHLO or HELO command needs to be sent."""

@property
def supported_auth_methods(self) -> list[str]:
    """List of authentication methods supported by server."""

def supports_extension(self, extension: str, /) -> bool:
    """Check if server supports a specific ESMTP extension."""

def get_transport_info(self, key: str) -> Any:
    """Get transport information (SSL info, socket details, etc.)."""

Email Sending Methods

Send emails using the SMTP client with different message formats.

async def sendmail(
    self,
    sender: str,
    recipients: Union[str, Sequence[str]],
    message: Union[str, bytes],
    *,
    mail_options: Optional[Sequence[str]] = None,
    rcpt_options: Optional[Sequence[str]] = None,
) -> tuple[dict[str, SMTPResponse], str]:
    """
    Send raw email message.
    
    Parameters:
    - sender: From email address
    - recipients: Recipient email addresses
    - message: Raw message content
    - mail_options: Options for MAIL command
    - rcpt_options: Options for RCPT command
    
    Returns:
    Tuple of (recipient_responses, data_response)
    """

async def send_message(
    self,
    message: Union[EmailMessage, Message],
    *,
    sender: Optional[str] = None,
    recipients: Optional[Union[str, Sequence[str]]] = None,
    mail_options: Optional[Sequence[str]] = None,
    rcpt_options: Optional[Sequence[str]] = None,
) -> tuple[dict[str, SMTPResponse], str]:
    """
    Send EmailMessage or Message object.
    
    Automatically extracts sender and recipients from message headers
    if not provided explicitly.
    """

def sendmail_sync(
    self,
    sender: str,
    recipients: Union[str, Sequence[str]],
    message: Union[str, bytes],
    *,
    mail_options: Optional[Sequence[str]] = None,
    rcpt_options: Optional[Sequence[str]] = None,
) -> tuple[dict[str, SMTPResponse], str]:
    """Synchronous wrapper for sendmail() using asyncio.run()."""

def send_message_sync(
    self,
    message: Union[EmailMessage, Message],
    *,
    sender: Optional[str] = None,
    recipients: Optional[Union[str, Sequence[str]]] = None,
    mail_options: Optional[Sequence[str]] = None,
    rcpt_options: Optional[Sequence[str]] = None,
) -> tuple[dict[str, SMTPResponse], str]:
    """Synchronous wrapper for send_message() using asyncio.run()."""

SMTP Protocol Commands

Low-level SMTP command methods for advanced use cases.

async def execute_command(
    self,
    command: Union[str, bytes],
    timeout: Optional[float] = None,
) -> SMTPResponse:
    """Execute arbitrary SMTP command and return response."""

async def helo(self, name: Optional[str] = None) -> SMTPResponse:
    """Send HELO command with local hostname."""

async def ehlo(self, name: Optional[str] = None) -> SMTPResponse:
    """Send EHLO command with local hostname."""

async def help(self, command: Optional[str] = None) -> SMTPResponse:
    """Send HELP command, optionally for specific command."""

async def rset(self) -> SMTPResponse:
    """Send RSET command to reset session state."""

async def noop(self) -> SMTPResponse:
    """Send NOOP command (no operation)."""

async def vrfy(self, address: str) -> SMTPResponse:
    """Send VRFY command to verify email address."""

async def expn(self, address: str) -> SMTPResponse:
    """Send EXPN command to expand mailing list."""

async def mail(
    self,
    sender: str,
    options: Optional[Sequence[str]] = None,
) -> SMTPResponse:
    """Send MAIL FROM command."""

async def rcpt(
    self,
    recipient: str,
    options: Optional[Sequence[str]] = None,
) -> SMTPResponse:
    """Send RCPT TO command."""

async def data(self, message: Union[str, bytes]) -> SMTPResponse:
    """Send DATA command with message content."""

TLS and Authentication

Handle TLS encryption and authentication operations.

async def starttls(
    self,
    tls_context: Optional[ssl.SSLContext] = None,
    *,
    validate_certs: bool = True,
    client_cert: Optional[str] = None,
    client_key: Optional[str] = None,
    cert_bundle: Optional[str] = None,
) -> SMTPResponse:
    """Initiate STARTTLS encryption upgrade."""

async def login(
    self,
    username: Union[str, bytes],
    password: Union[str, bytes],
    *,
    method: Optional[str] = None,
) -> SMTPResponse:
    """Authenticate with server using best available method."""

async def auth_crammd5(
    self,
    username: Union[str, bytes],
    password: Union[str, bytes],
) -> SMTPResponse:
    """Authenticate using CRAM-MD5 method."""

async def auth_plain(
    self,
    username: Union[str, bytes],
    password: Union[str, bytes],
) -> SMTPResponse:
    """Authenticate using PLAIN method."""

async def auth_login(
    self,
    username: Union[str, bytes],
    password: Union[str, bytes],
) -> SMTPResponse:
    """Authenticate using LOGIN method."""

Usage Examples

Basic Client Usage

import asyncio
import aiosmtplib
from email.message import EmailMessage

async def basic_client_usage():
    # Create client instance
    smtp = aiosmtplib.SMTP(hostname="localhost", port=1025)
    
    try:
        # Connect to server
        await smtp.connect()
        print(f"Connected: {smtp.is_connected}")
        
        # Send email
        message = EmailMessage()
        message["From"] = "sender@example.com"
        message["To"] = "recipient@example.com"
        message["Subject"] = "Test Email"
        message.set_content("Hello from SMTP client!")
        
        response = await smtp.send_message(message)
        print(f"Email sent: {response}")
        
    finally:
        # Clean up
        smtp.close()

asyncio.run(basic_client_usage())

Context Manager Usage

import asyncio
import aiosmtplib

async def context_manager_usage():
    # Automatic connection management
    async with aiosmtplib.SMTP(hostname="localhost", port=1025) as smtp:
        print(f"Connected: {smtp.is_connected}")
        
        # Send multiple emails in same session
        for i in range(3):
            response = await smtp.sendmail(
                "sender@example.com",
                ["recipient@example.com"],
                f"Subject: Message {i}\n\nThis is message {i}"
            )
            print(f"Message {i} sent: {response[1]}")

asyncio.run(context_manager_usage())

Advanced Configuration

import asyncio
import ssl
import aiosmtplib

async def advanced_configuration():
    # Custom SSL context
    context = ssl.create_default_context()
    context.check_hostname = False
    
    # Create client with advanced options
    smtp = aiosmtplib.SMTP(
        hostname="smtp.gmail.com",
        port=587,
        username="your-email@gmail.com",
        password="your-password",
        start_tls=True,
        validate_certs=True,
        tls_context=context,
        timeout=30,
        local_hostname="my.local.host"
    )
    
    async with smtp:
        # Check server capabilities
        print(f"Supported auth methods: {smtp.supported_auth_methods}")
        print(f"Supports PIPELINING: {smtp.supports_extension('PIPELINING')}")
        print(f"Last EHLO response: {smtp.last_ehlo_response}")
        
        # Send email with custom options
        response = await smtp.sendmail(
            "sender@gmail.com",
            ["recipient@example.com"],
            "Subject: Advanced Config\n\nEmail with advanced configuration",
            mail_options=["BODY=8BITMIME"],
            rcpt_options=["NOTIFY=SUCCESS,FAILURE"]
        )
        print(f"Email sent with options: {response}")

asyncio.run(advanced_configuration())

Error Handling and Recovery

import asyncio
import aiosmtplib

async def error_handling_example():
    smtp = aiosmtplib.SMTP(hostname="smtp.example.com", port=587)
    
    try:
        await smtp.connect()
        
        # Attempt authentication
        try:
            await smtp.login("user", "password")
        except aiosmtplib.SMTPAuthenticationError:
            print("Authentication failed, trying different credentials")
            await smtp.login("user", "different_password")
        
        # Send email with error handling
        try:
            response = await smtp.sendmail(
                "sender@example.com",
                ["invalid@nonexistent.domain"],
                "Subject: Test\n\nTest message"
            )
            
            # Check individual recipient responses
            for recipient, smtp_response in response[0].items():
                if smtp_response.code >= 400:
                    print(f"Failed to send to {recipient}: {smtp_response}")
                else:
                    print(f"Successfully sent to {recipient}")
                    
        except aiosmtplib.SMTPRecipientsRefused as e:
            print(f"All recipients refused: {e.recipients}")
        except aiosmtplib.SMTPSenderRefused as e:
            print(f"Sender refused: {e.sender}")
            
    except aiosmtplib.SMTPConnectError as e:
        print(f"Connection failed: {e}")
    except aiosmtplib.SMTPTimeoutError as e:
        print(f"Operation timed out: {e}")
    finally:
        if smtp.is_connected:
            await smtp.quit()

asyncio.run(error_handling_example())

Low-Level Command Usage

import asyncio
import aiosmtplib

async def low_level_commands():
    smtp = aiosmtplib.SMTP(hostname="localhost", port=1025)
    
    async with smtp:
        # Manual SMTP conversation
        ehlo_response = await smtp.ehlo("my.domain.com")
        print(f"EHLO response: {ehlo_response}")
        
        # Check if authentication is needed
        if smtp.supported_auth_methods:
            auth_response = await smtp.login("user", "pass")
            print(f"AUTH response: {auth_response}")
        
        # Manual mail transaction
        mail_response = await smtp.mail("sender@example.com")
        print(f"MAIL response: {mail_response}")
        
        rcpt_response = await smtp.rcpt("recipient@example.com")
        print(f"RCPT response: {rcpt_response}")
        
        data_response = await smtp.data("Subject: Manual\n\nManual message")
        print(f"DATA response: {data_response}")
        
        # Reset for next transaction
        rset_response = await smtp.rset()
        print(f"RSET response: {rset_response}")

asyncio.run(low_level_commands())

Class Constants

class SMTP:
    # Preferred authentication methods in order
    AUTH_METHODS: tuple[str, ...] = ("cram-md5", "plain", "login")

# Module-level constants
SMTP_PORT = 25
SMTP_TLS_PORT = 465
SMTP_STARTTLS_PORT = 587
DEFAULT_TIMEOUT = 60

Install with Tessl CLI