🔍 Code Extractor

SMTPServer is a class that manages an SMTP server for receiving emails, handling their forwarding, and providing server statistics.

Purpose

This class provides a complete SMTP server implementation using aiosmtpd for receiving emails. It manages the server lifecycle (start/stop), handles incoming emails through an EmailForwardingHandler, supports graceful shutdown with signal handling, and provides statistics about server operations. It's designed to be used as a standalone email receiving service with configurable host, port, and message size limits.

Source Code

class SMTPServer:
    """SMTP Server for receiving emails."""
    
    def __init__(self, 
                 host: str = None,
                 port: int = None,
                 max_message_size: int = None):
        """Initialize the SMTP server."""
        self.host = host or settings.SMTP_LISTEN_HOST
        self.port = port or settings.SMTP_LISTEN_PORT
        self.max_message_size = max_message_size or settings.MAX_MESSAGE_SIZE
        
        self.handler = EmailForwardingHandler()
        self.controller = None
        self.running = False
        
    def start(self):
        """Start the SMTP server."""
        try:
            settings.validate_config()
            
            self.controller = Controller(
                self.handler,
                hostname=self.host,
                port=self.port,
                data_size_limit=self.max_message_size
            )
            
            self.controller.start()
            self.running = True
            
            logger.info(f"SMTP server started on {self.host}:{self.port}")
            logger.info(f"Max message size: {self.max_message_size} bytes")
            
        except Exception as e:
            logger.error(f"Failed to start SMTP server: {e}")
            raise
            
    def stop(self):
        """Stop the SMTP server."""
        if self.controller:
            self.controller.stop()
            self.running = False
            logger.info("SMTP server stopped")
            
    def get_stats(self):
        """Get server statistics."""
        stats = self.handler.server_stats.copy()
        stats.update(self.handler.email_handler.get_stats())
        return stats
        
    async def run_forever(self):
        """Run the server forever with graceful shutdown handling."""
        import signal
        
        def signal_handler(signum, frame):
            logger.info(f"Received signal {signum}, shutting down...")
            self.stop()
            
        # Register signal handlers
        signal.signal(signal.SIGINT, signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)
        
        self.start()
        
        try:
            while self.running:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            logger.info("Received keyboard interrupt")
        finally:
            self.stop()

Parameters

Parameter Details

host: The hostname or IP address where the SMTP server will listen for connections. If None, defaults to settings.SMTP_LISTEN_HOST. Typically '0.0.0.0' for all interfaces or 'localhost' for local only.

port: The port number on which the SMTP server will listen. If None, defaults to settings.SMTP_LISTEN_PORT. Standard SMTP ports are 25, 587, or custom ports like 1025 for development.

max_message_size: Maximum size in bytes for incoming email messages. If None, defaults to settings.MAX_MESSAGE_SIZE. This prevents memory exhaustion from extremely large emails.

Return Value

Instantiation returns an SMTPServer object. The start() method returns None but starts the server. The stop() method returns None and stops the server. The get_stats() method returns a dictionary containing server statistics including message counts and handler statistics. The run_forever() method is an async coroutine that returns None and runs until interrupted.

Class Interface

Methods

Attributes

Dependencies

• asyncio
• logging
• aiosmtpd
• email
• typing
• signal
• sys
• os

Required Imports

import asyncio
import logging
from aiosmtpd.controller import Controller
from email.message import EmailMessage
from typing import Optional
from email_handler import EmailHandler
from forwarder.email_handler import EmailHandler
from config import settings
import signal
import sys
import os

Conditional/Optional Imports

These imports are only needed under specific conditions:

import signal

Usage Example

import asyncio
from smtp_server import SMTPServer

# Basic instantiation with defaults from settings
server = SMTPServer()

# Or with custom configuration
server = SMTPServer(host='0.0.0.0', port=1025, max_message_size=10485760)

# Start the server
server.start()

# Check if running
if server.running:
    print('Server is running')

# Get statistics
stats = server.get_stats()
print(f'Server stats: {stats}')

# Stop the server
server.stop()

# Or run forever with graceful shutdown
async def main():
    server = SMTPServer()
    await server.run_forever()

if __name__ == '__main__':
    asyncio.run(main())

Best Practices

• Always call start() before using the server to receive emails
• Always call stop() when done to properly clean up resources and close connections
• Use run_forever() for long-running server processes as it includes proper signal handling for graceful shutdown
• Check the 'running' attribute to verify server state before performing operations
• Handle exceptions from start() as it may raise if configuration is invalid or port is already in use
• Call get_stats() periodically to monitor server health and email processing metrics
• Ensure settings.validate_config() passes before starting the server
• The server uses an EmailForwardingHandler internally which must be properly configured
• For production use, ensure proper logging configuration is in place before instantiation
• Signal handlers (SIGINT, SIGTERM) are only registered when using run_forever(), not with manual start()/stop()
• The controller attribute is None until start() is called successfully

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class EmailForwardingHandler 73.9% similar

  SMTP message handler class that processes incoming email messages and forwards them using an EmailHandler instance while tracking server statistics.

  From: /tf/active/vicechatdev/email-forwarder/src/forwarder/smtp_server.py

• function run_smtp_server 73.0% similar

  Initializes and runs an asynchronous SMTP server using a new event loop, with proper error handling and graceful shutdown.

  From: /tf/active/vicechatdev/email-forwarder/src/forwarder/smtp_server.py

• class TestSmtpServer 71.9% similar

  A test class for validating the functionality of the SmtpServer class, including server startup, email handling, email forwarding, and error handling for invalid inputs.

  From: /tf/active/vicechatdev/email-forwarder/tests/test_smtp_server.py

• function run_server 64.2% similar

  Initializes and runs an asynchronous SMTP server with platform-specific event loop configuration and error handling.

  From: /tf/active/vicechatdev/email-forwarder/src/main.py

• function main_v9 61.9% similar

  Asynchronous main entry point function that initializes and runs an email forwarding SMTP server with logging, configuration validation, and graceful shutdown handling.

  From: /tf/active/vicechatdev/email-forwarder/src/main.py