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

Utility Functions

Name: tessl/pypi-aiosmtplib
Author: tessl

aiosmtplib provides utility functions for authentication, email message processing, and ESMTP extension handling. These functions are useful for custom implementations, debugging, and applications that need fine-grained control over SMTP operations.

Capabilities

Authentication Utilities

Functions for encoding authentication credentials in various SMTP authentication protocols.

from aiosmtplib.auth import auth_plain_encode, auth_login_encode, auth_crammd5_verify

def auth_plain_encode(username: str, password: str) -> bytes:
    """
    Encode credentials for PLAIN authentication method.
    
    Parameters:
    - username: Authentication username
    - password: Authentication password
    
    Returns:
    Encoded credentials as bytes suitable for PLAIN AUTH command
    """

def auth_login_encode(username: str, password: str) -> tuple[bytes, bytes]:
    """
    Encode credentials for LOGIN authentication method.
    
    Parameters:
    - username: Authentication username  
    - password: Authentication password
    
    Returns:
    Tuple of (encoded_username, encoded_password) as bytes
    """

def auth_crammd5_verify(username: str, password: str, challenge: bytes) -> bytes:
    """
    Generate CRAM-MD5 authentication response.
    
    Parameters:
    - username: Authentication username
    - password: Authentication password
    - challenge: Challenge bytes from server
    
    Returns:
    CRAM-MD5 response bytes
    """

Email Message Utilities

Functions for processing email messages and extracting address information.

from aiosmtplib.email import extract_recipients, extract_sender, flatten_message, parse_address, quote_address

def extract_recipients(message: Union[EmailMessage, Message]) -> list[str]:
    """
    Extract all recipient addresses from email message headers.
    
    Processes To, Cc, and Bcc headers to build complete recipient list.
    
    Parameters:
    - message: EmailMessage or Message object
    
    Returns:
    List of recipient email addresses
    """

def extract_sender(message: Union[EmailMessage, Message]) -> Optional[str]:
    """
    Extract sender address from email message headers.
    
    Checks From header for sender address.
    
    Parameters:
    - message: EmailMessage or Message object
    
    Returns:
    Sender email address or None if not found
    """

def flatten_message(
    message: Union[EmailMessage, Message],
    *,
    utf8: bool = False,
    cte_type: str = "8bit"
) -> bytes:
    """
    Convert email message to bytes for SMTP transmission.
    
    Parameters:
    - message: EmailMessage or Message object
    - utf8: Use UTF-8 encoding
    - cte_type: Content transfer encoding type
    
    Returns:
    Message as bytes ready for SMTP DATA command
    """

def parse_address(address: str) -> str:
    """
    Parse and normalize email address string.
    
    Handles various email address formats and extracts the actual
    email address from display name formats.
    
    Parameters:
    - address: Email address string (may include display name)
    
    Returns:
    Normalized email address
    """

def quote_address(address: str) -> str:
    """
    Quote email address according to RFC 821 rules.
    
    Adds angle brackets around address if needed for SMTP commands.
    
    Parameters:
    - address: Email address string
    
    Returns:
    Properly quoted address for SMTP commands
    """

ESMTP Extension Utilities

Functions for parsing and handling ESMTP server extensions.

from aiosmtplib.esmtp import parse_esmtp_extensions

def parse_esmtp_extensions(message: str) -> tuple[dict[str, str], list[str]]:
    """
    Parse EHLO response into extensions dictionary and auth methods list.
    
    Parameters:
    - message: Raw EHLO response message from server
    
    Returns:
    Tuple of (extensions_dict, auth_methods_list) where:
    - extensions_dict maps extension names to their parameters
    - auth_methods_list contains supported authentication methods
    """

Low-Level Protocol Access

Advanced protocol class for custom SMTP implementations.

class SMTPProtocol:
    """
    Low-level asyncio protocol for SMTP communication.
    
    Advanced users can use this for custom SMTP implementations
    that need direct protocol access.
    """
    def __init__(
        self,
        loop: Optional[asyncio.AbstractEventLoop] = None,
        connection_lost_callback: Optional[Callable[[], None]] = None,
    ) -> None: ...

Usage Examples

Authentication Utilities

from aiosmtplib.auth import auth_plain_encode, auth_login_encode, auth_crammd5_verify

# Encode credentials for different auth methods
username = "user@example.com"
password = "secret123"

# PLAIN authentication
plain_creds = auth_plain_encode(username, password)
print(f"PLAIN credentials: {plain_creds}")

# LOGIN authentication  
login_user, login_pass = auth_login_encode(username, password)
print(f"LOGIN username: {login_user}")
print(f"LOGIN password: {login_pass}")

# CRAM-MD5 authentication (requires server challenge)
challenge = b"<12345.678@example.com>"
crammd5_response = auth_crammd5_verify(username, password, challenge)
print(f"CRAM-MD5 response: {crammd5_response}")

Email Message Processing

from aiosmtplib.email import extract_recipients, extract_sender, flatten_message, parse_address, quote_address
from email.message import EmailMessage

# Create test message
message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient1@example.com, recipient2@example.com"
message["Cc"] = "cc@example.com"
message["Bcc"] = "bcc@example.com"
message["Subject"] = "Test Message"
message.set_content("This is a test message.")

# Extract recipients
recipients = extract_recipients(message)
print(f"All recipients: {recipients}")
# Output: ['recipient1@example.com', 'recipient2@example.com', 'cc@example.com', 'bcc@example.com']

# Extract sender
sender = extract_sender(message)
print(f"Sender: {sender}")
# Output: sender@example.com

# Convert to bytes for transmission
message_bytes = flatten_message(message)
print(f"Message size: {len(message_bytes)} bytes")

# Parse and quote addresses
raw_address = "John Doe <john@example.com>"
parsed = parse_address(raw_address)
quoted = quote_address(parsed)
print(f"Raw: {raw_address}")
print(f"Parsed: {parsed}")
print(f"Quoted: {quoted}")

ESMTP Extension Parsing

from aiosmtplib.esmtp import parse_esmtp_extensions

# Example EHLO response from server
ehlo_response = """250-smtp.example.com Hello [192.168.1.100]
250-SIZE 35882577
250-8BITMIME
250-PIPELINING
250-CHUNKING
250-SMTPUTF8
250-AUTH PLAIN LOGIN CRAM-MD5
250-AUTH=PLAIN LOGIN CRAM-MD5
250 HELP"""

# Parse extensions and auth methods
extensions, auth_methods = parse_esmtp_extensions(ehlo_response)

print("Server extensions:")
for ext_name, ext_value in extensions.items():
    print(f"  {ext_name}: {ext_value}")

print(f"\nSupported auth methods: {auth_methods}")

# Check for specific extensions
if "PIPELINING" in extensions:
    print("Server supports command pipelining")

if "SIZE" in extensions:
    max_size = extensions["SIZE"]
    print(f"Maximum message size: {max_size} bytes")

Custom Authentication Implementation

import asyncio
import aiosmtplib
import base64

async def custom_auth_example():
    smtp = aiosmtplib.SMTP(hostname="localhost", port=1025)
    
    try:
        await smtp.connect()
        
        # Check what auth methods server supports
        auth_methods = smtp.supported_auth_methods
        print(f"Server supports: {auth_methods}")
        
        if "PLAIN" in auth_methods:
            # Manual PLAIN authentication
            credentials = aiosmtplib.auth_plain_encode("user", "pass")
            auth_string = base64.b64encode(credentials).decode()
            
            # Send AUTH command manually
            response = await smtp.execute_command(f"AUTH PLAIN {auth_string}")
            print(f"Manual PLAIN auth response: {response}")
            
        elif "LOGIN" in auth_methods:
            # Manual LOGIN authentication
            username_b64, password_b64 = aiosmtplib.auth_login_encode("user", "pass")
            
            # LOGIN authentication is interactive
            response1 = await smtp.execute_command("AUTH LOGIN")
            print(f"AUTH LOGIN response: {response1}")
            
            response2 = await smtp.execute_command(username_b64.decode())
            print(f"Username response: {response2}")
            
            response3 = await smtp.execute_command(password_b64.decode())
            print(f"Password response: {response3}")
        
    finally:
        smtp.close()

asyncio.run(custom_auth_example())

Message Validation and Processing

import aiosmtplib
from email.message import EmailMessage
import email.utils

def validate_and_process_message(message):
    """Validate and process email message before sending."""
    
    # Extract and validate sender
    sender = aiosmtplib.extract_sender(message)
    if not sender:
        raise ValueError("Message must have a From header")
    
    # Validate sender format
    try:
        parsed_sender = aiosmtplib.parse_address(sender)
        if "@" not in parsed_sender:
            raise ValueError(f"Invalid sender address: {sender}")
    except Exception as e:
        raise ValueError(f"Cannot parse sender address: {e}")
    
    # Extract and validate recipients
    recipients = aiosmtplib.extract_recipients(message)
    if not recipients:
        raise ValueError("Message must have recipients (To, Cc, or Bcc)")
    
    # Validate each recipient
    valid_recipients = []
    for recipient in recipients:
        try:
            parsed_recipient = aiosmtplib.parse_address(recipient)
            if "@" not in parsed_recipient:
                print(f"Warning: Skipping invalid recipient: {recipient}")
                continue
            valid_recipients.append(parsed_recipient)
        except Exception as e:
            print(f"Warning: Cannot parse recipient {recipient}: {e}")
    
    if not valid_recipients:
        raise ValueError("No valid recipients found")
    
    # Add Message-ID if missing
    if "Message-ID" not in message:
        message["Message-ID"] = email.utils.make_msgid()
    
    # Add Date if missing
    if "Date" not in message:
        message["Date"] = email.utils.formatdate(localtime=True)
    
    return parsed_sender, valid_recipients

# Example usage
message = EmailMessage()
message["From"] = "John Doe <john@example.com>"
message["To"] = "recipient@example.com, invalid-email"
message["Subject"] = "Test Message"
message.set_content("This is a test message.")

try:
    sender, recipients = validate_and_process_message(message)
    print(f"Validated sender: {sender}")
    print(f"Valid recipients: {recipients}")
    print(f"Message-ID: {message['Message-ID']}")
    print(f"Date: {message['Date']}")
except ValueError as e:
    print(f"Message validation failed: {e}")

Advanced Message Processing

import aiosmtplib
from email.message import EmailMessage
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
import email.encoders

def create_complex_message():
    """Create a complex multipart message."""
    
    # Create multipart message
    msg = MIMEMultipart("alternative")
    msg["From"] = "sender@example.com"
    msg["To"] = "recipient@example.com"
    msg["Subject"] = "Complex Message"
    
    # Add text part
    text_part = MIMEText("This is the plain text version.", "plain")
    msg.attach(text_part)
    
    # Add HTML part  
    html_part = MIMEText("<h1>This is the HTML version</h1>", "html")
    msg.attach(html_part)
    
    return msg

async def send_complex_message():
    # Create complex message
    message = create_complex_message()
    
    # Process with utilities
    sender = aiosmtplib.extract_sender(message)
    recipients = aiosmtplib.extract_recipients(message)
    message_bytes = aiosmtplib.flatten_message(message, utf8=True)
    
    print(f"Sender: {sender}")
    print(f"Recipients: {recipients}")
    print(f"Message size: {len(message_bytes)} bytes")
    
    # Send using SMTP client
    smtp = aiosmtplib.SMTP(hostname="localhost", port=1025)
    async with smtp:
        response = await smtp.send_message(message)
        print(f"Message sent: {response}")

asyncio.run(send_complex_message())

Internal vs Public Utilities

Public Utilities (Exported in all)

These functions are intended for public use:

Authentication encoding functions
Email message processing functions
ESMTP extension parsing
SMTPProtocol for advanced use cases

Internal Utilities (Not Recommended for Direct Use)

Some utilities are for internal use but may be useful in specific scenarios:

Low-level protocol functions
Internal message parsing helpers
Connection management utilities

When using any utility functions, prefer the high-level send() function or SMTP class for most use cases, and only use utilities when you need specific functionality not provided by the main API.

Install with Tessl CLI