tessl/pypi-twisted
An asynchronous networking framework written in Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
email.mddocs/
Email Protocols
Complete email protocol implementations including SMTP, POP3, and IMAP4 with support for both client and server operations. Twisted provides comprehensive email handling capabilities.
Capabilities
SMTP (Simple Mail Transfer Protocol)
SMTP server and client implementations for sending and receiving email.
class smtp.SMTP:
"""
SMTP server protocol implementation.
Attributes:
- host: Server hostname
- portal: Authentication portal
- challenger: Authentication challenger
"""
host = None
portal = None
challenger = None
def greeting(self):
"""
Get server greeting message.
Returns:
str: SMTP greeting
"""
def extensions(self):
"""
Get supported SMTP extensions.
Returns:
dict: Extension capabilities
"""
def validateTo(self, user):
"""
Validate recipient address.
Args:
user: User address object
Returns:
callable: Message delivery callable
"""
def validateFrom(self, helo, origin):
"""
Validate sender address.
Args:
helo (bytes): HELO/EHLO string
origin: Sender address object
Returns:
Deferred: Validation result
"""
class smtp.ESMTP(smtp.SMTP):
"""
Extended SMTP server with additional features.
Supports SMTP extensions like authentication, TLS, etc.
"""
def ext_AUTH(self, args):
"""Handle AUTH extension."""
def ext_STARTTLS(self, args):
"""Handle STARTTLS extension."""
def ext_SIZE(self, args):
"""Handle SIZE extension."""
class smtp.ESMTPSender:
"""
SMTP client for sending email messages.
"""
def __init__(self, username=None, password=None, contextFactory=None, **kwargs):
"""
Args:
username (bytes): SMTP username
password (bytes): SMTP password
contextFactory: SSL context factory for STARTTLS
**kwargs: Additional connection parameters
"""
def sendmail(self, from_addr, to_addrs, msg, response=None):
"""
Send email message.
Args:
from_addr (str): Sender address
to_addrs (list): Recipient addresses
msg (bytes): Email message content
response: Response handler
Returns:
Deferred: Delivery result
"""
class smtp.SMTPSenderFactory:
"""
Factory for creating SMTP sender connections.
"""
def __init__(self, fromEmail, toEmail, file, deferred, retries=5, timeout=None):
"""
Args:
fromEmail (str): Sender address
toEmail (str): Recipient address
file: File-like object with message content
deferred: Deferred to fire when complete
retries (int): Number of retry attempts
timeout (int): Connection timeout
"""
def buildProtocol(self, addr):
"""
Build SMTP sender protocol.
Args:
addr: Connection address
Returns:
ESMTPSender: SMTP sender instance
"""
class smtp.SMTPFactory:
"""
SMTP server factory.
"""
def __init__(self):
self.deliveryClass = smtp.SMTPDelivery
self.portal = None
self.challenger = None
def buildProtocol(self, addr):
"""
Build SMTP server protocol.
Args:
addr: Connection address
Returns:
SMTP: SMTP server instance
"""SMTP Usage Example:
from twisted.mail import smtp
from twisted.internet import reactor, defer, endpoints
from twisted.cred import portal, checkers
# SMTP Client - Send email
@defer.inlineCallbacks
def send_email():
sender_factory = smtp.SMTPSenderFactory(
fromEmail="sender@example.com",
toEmail="recipient@example.com",
file=io.StringIO("Subject: Test\n\nHello, World!"),
deferred=defer.Deferred()
)
endpoint = endpoints.TCP4ClientEndpoint(reactor, "smtp.example.com", 25)
yield endpoint.connect(sender_factory)
# SMTP Server
class MyDelivery:
"""Custom message delivery handler."""
def receivedHeader(self, helo, origin, recipients):
return "Received: from %s" % (helo,)
def validateTo(self, user):
# Validate recipient
if user.dest.local == "postmaster":
return lambda: MyMessage()
raise smtp.SMTPBadRcpt(user)
def validateFrom(self, helo, origin):
# Validate sender
return origin
class MyMessage:
"""Message storage handler."""
def eomReceived(self):
# Called when message is complete
print("Message received")
return defer.succeed("250 Message accepted")
# Set up SMTP server
factory = smtp.SMTPFactory()
factory.delivery = MyDelivery()
factory.portal = portal.Portal(realm, [checker])
endpoint = endpoints.TCP4ServerEndpoint(reactor, 25)
endpoint.listen(factory)POP3 (Post Office Protocol)
POP3 server and client for retrieving email from mailboxes.
class pop3.POP3:
"""
POP3 server protocol implementation.
Attributes:
- mbox: Mailbox object
- portal: Authentication portal
"""
mbox = None
portal = None
def authenticateUserAPOP(self, user, digest):
"""
Authenticate using APOP.
Args:
user (bytes): Username
digest (bytes): APOP digest
Returns:
Deferred: Authentication result
"""
def authenticateUserPASS(self, user, password):
"""
Authenticate using USER/PASS.
Args:
user (bytes): Username
password (bytes): Password
Returns:
Deferred: Authentication result
"""
class pop3.POP3Client:
"""
POP3 client protocol implementation.
"""
def login(self, username, password):
"""
Login to POP3 server.
Args:
username (bytes): Username
password (bytes): Password
Returns:
Deferred: Login result
"""
def listMessages(self):
"""
List messages in mailbox.
Returns:
Deferred[list]: List of (message_num, size) tuples
"""
def getMessage(self, index):
"""
Retrieve message by index.
Args:
index (int): Message index
Returns:
Deferred[list]: Message lines
"""
def deleteMessage(self, index):
"""
Mark message for deletion.
Args:
index (int): Message index
Returns:
Deferred: Deletion result
"""
def quit(self):
"""
Quit POP3 session.
Returns:
Deferred: Quit result
"""
class pop3.POP3Factory:
"""
POP3 server factory.
"""
def __init__(self):
self.portal = None
def buildProtocol(self, addr):
"""
Build POP3 server protocol.
Args:
addr: Connection address
Returns:
POP3: POP3 server instance
"""
class pop3.POP3ClientFactory:
"""
POP3 client factory.
"""
def __init__(self, username, password, **kwargs):
"""
Args:
username (str): POP3 username
password (str): POP3 password
**kwargs: Additional parameters
"""POP3 Usage Example:
from twisted.mail import pop3
from twisted.internet import reactor, defer, endpoints
# POP3 Client
@defer.inlineCallbacks
def retrieve_mail():
factory = pop3.POP3ClientFactory("username", "password")
endpoint = endpoints.TCP4ClientEndpoint(reactor, "pop.example.com", 110)
protocol = yield endpoint.connect(factory)
yield protocol.login(b"username", b"password")
messages = yield protocol.listMessages()
print(f"Found {len(messages)} messages")
# Retrieve first message
if messages:
message_lines = yield protocol.getMessage(0)
message = b'\n'.join(message_lines)
print(f"Message: {message.decode()}")
yield protocol.quit()
# POP3 Server with custom mailbox
class MyMailbox:
"""Custom mailbox implementation."""
def __init__(self, messages):
self.messages = messages
def listMessages(self):
return [(i, len(msg)) for i, msg in enumerate(self.messages)]
def getMessage(self, i):
return self.messages[i].split(b'\n')
def deleteMessage(self, i):
del self.messages[i]
class MyRealm:
def requestAvatar(self, avatarId, mind, *interfaces):
# Create mailbox for user
messages = [b"Subject: Test\n\nHello!"]
mailbox = MyMailbox(messages)
return (pop3.IMailbox, mailbox, lambda: None)
# Set up POP3 server
realm = MyRealm()
checker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
checker.addUser(b"testuser", b"password")
factory = pop3.POP3Factory()
factory.portal = portal.Portal(realm, [checker])
endpoint = endpoints.TCP4ServerEndpoint(reactor, 110)
endpoint.listen(factory)IMAP4 (Internet Message Access Protocol)
IMAP4 server and client for advanced email access and management.
class imap4.IMAP4Server:
"""
IMAP4 server protocol implementation.
Attributes:
- portal: Authentication portal
- account: User account object
"""
portal = None
account = None
def authenticateLogin(self, username, password):
"""
Authenticate with LOGIN command.
Args:
username (bytes): Username
password (bytes): Password
Returns:
Deferred: Authentication result
"""
def do_SELECT(self, tag, messages):
"""Handle SELECT command."""
def do_EXAMINE(self, tag, messages):
"""Handle EXAMINE command."""
def do_FETCH(self, tag, messages):
"""Handle FETCH command."""
def do_SEARCH(self, tag, messages):
"""Handle SEARCH command."""
def do_STORE(self, tag, messages):
"""Handle STORE command."""
class imap4.IMAP4Client:
"""
IMAP4 client protocol implementation.
"""
def login(self, username, password):
"""
Login to IMAP4 server.
Args:
username (str): Username
password (str): Password
Returns:
Deferred: Login result
"""
def list(self, reference, name):
"""
List mailboxes.
Args:
reference (str): Reference name
name (str): Mailbox name pattern
Returns:
Deferred[list]: List of mailboxes
"""
def select(self, mailbox):
"""
Select mailbox.
Args:
mailbox (str): Mailbox name
Returns:
Deferred[dict]: Mailbox status
"""
def search(self, *query):
"""
Search for messages.
Args:
*query: Search criteria
Returns:
Deferred[list]: Message numbers
"""
def fetch(self, messages, *query):
"""
Fetch message data.
Args:
messages: Message set
*query: Fetch items
Returns:
Deferred[dict]: Message data
"""
def store(self, messages, command, flags):
"""
Store message flags.
Args:
messages: Message set
command (str): Store command
flags (list): Message flags
Returns:
Deferred: Store result
"""
def expunge(self):
"""
Expunge deleted messages.
Returns:
Deferred[list]: Expunged message numbers
"""
class imap4.IMAP4Factory:
"""
IMAP4 server factory.
"""
def __init__(self):
self.portal = None
self.timeOut = 30 * 60 # 30 minutes
def buildProtocol(self, addr):
"""
Build IMAP4 server protocol.
Args:
addr: Connection address
Returns:
IMAP4Server: IMAP4 server instance
"""
class imap4.IMAP4ClientFactory:
"""
IMAP4 client factory.
"""
def __init__(self, username, password, **kwargs):
"""
Args:
username (str): IMAP username
password (str): IMAP password
**kwargs: Additional parameters
"""Email Message Handling
Utilities for parsing, creating, and manipulating email messages.
class mail.MailService:
"""
Service for handling email delivery.
Attributes:
- domains: List of handled domains
- portals: Authentication portals
"""
domains = None
portals = None
def addDomain(self, domain):
"""
Add domain to handle.
Args:
domain (str): Domain name
"""
def setQueue(self, queue):
"""
Set message queue.
Args:
queue: Queue object for messages
"""
class mail.FileMessage:
"""
File-based email message storage.
"""
def __init__(self, fp, name, finalName):
"""
Args:
fp: File pointer
name (str): Temporary filename
finalName (str): Final filename
"""
def eomReceived(self):
"""
Called when end of message is received.
Returns:
Deferred: Processing result
"""
# Email address parsing
class smtp.Address:
"""
Email address representation.
Attributes:
- local: Local part of address
- domain: Domain part of address
"""
local = None
domain = None
def __init__(self, local, domain=None):
"""
Args:
local (str): Local part
domain (str): Domain part
"""
def __str__(self):
"""String representation of address."""Maildir Support
Maildir format support for mail storage.
class maildir.MaildirMessage:
"""
Maildir format message.
"""
def __init__(self, file, flags=None):
"""
Args:
file: Message file
flags (str): Message flags
"""
class maildir.AbstractMaildirDomain:
"""
Abstract Maildir domain handler.
"""
def __init__(self, root):
"""
Args:
root (str): Maildir root directory
"""
def userDirectory(self, user):
"""
Get user's maildir directory.
Args:
user (str): Username
Returns:
str: User directory path
"""Email Message Example:
from twisted.mail import smtp, mail
from twisted.internet import defer
import email.mime.text
# Create and send email message
def create_message():
msg = email.mime.text.MIMEText("Hello, World!")
msg['From'] = "sender@example.com"
msg['To'] = "recipient@example.com"
msg['Subject'] = "Test Message"
return msg.as_bytes()
@defer.inlineCallbacks
def send_message():
message_data = create_message()
factory = smtp.SMTPSenderFactory(
fromEmail="sender@example.com",
toEmail="recipient@example.com",
file=io.BytesIO(message_data),
deferred=defer.Deferred()
)
endpoint = endpoints.TCP4ClientEndpoint(reactor, "smtp.example.com", 25)
yield endpoint.connect(factory)
print("Email sent successfully")
# Handle incoming email
class CustomDelivery:
def validateTo(self, user):
# Accept all messages
return lambda: CustomMessage(user)
def validateFrom(self, helo, origin):
return origin
class CustomMessage:
def __init__(self, user):
self.user = user
self.lines = []
def lineReceived(self, line):
self.lines.append(line)
def eomReceived(self):
# Process complete message
message = b'\n'.join(self.lines)
print(f"Received message for {self.user}: {message}")
return defer.succeed("250 Message accepted")Install with Tessl CLI
npx tessl i tessl/pypi-twisted