编程方式用法 ¶

If you already have an asyncio 事件循环 , you can create a server 使用 SMTP class as the protocol factory , and then run the loop forever. If you need to pass arguments to the SMTP constructor, use functools.partial() or write your own wrapper function. You might also want to add a signal handler so that the loop can be stopped, say when you hit control-C.

It’s probably easier to use a controller which runs the SMTP server in a separate thread with a dedicated event loop. The controller provides useful and reliable start and stop semantics so that the foreground thread doesn’t block. Among other use cases, this makes it convenient to spin up an SMTP server for unit tests.

In both cases, you need to pass a handler 到 SMTP constructor. Handlers respond to events that you care about during the SMTP dialog.

使用控制器 ¶

TCP-based Server ¶

Controller class creates a TCP-based server, listening on an Internet endpoint (i.e., ip_address:port pair).

Say you want to receive email for example.com and print incoming mail data to the console. Start by implementing a handler as follows:

>>> import asyncio
>>> class ExampleHandler:
...     async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
...         if not address.endswith('@example.com'):
...             return '550 not relaying to that domain'
...         envelope.rcpt_tos.append(address)
...         return '250 OK'
...
...     async def handle_DATA(self, server, session, envelope):
...         print('Message from %s' % envelope.mail_from)
...         print('Message for %s' % envelope.rcpt_tos)
...         print('Message data:\n')
...         for ln in envelope.content.decode('utf8', errors='replace').splitlines():
...             print(f'> {ln}'.strip())
...         print()
...         print('End of message')
...         return '250 Message accepted for delivery'

Pass an instance of your ExampleHandler class to the Controller , and then start it:

>>> from aiosmtpd.controller import Controller
>>> controller = Controller(ExampleHandler())
>>> controller.start()

The SMTP thread might run into errors during its setup phase; to catch this the main thread will timeout when waiting for the SMTP server to become ready. By default the timeout is set to 1 second but can be changed either by using the AIOSMTPD_CONTROLLER_TIMEOUT environment variable or by passing a different ready_timeout duration to the Controller’s constructor.

Connect to the server and send a message, which then gets printed by ExampleHandler :

>>> from smtplib import SMTP as Client
>>> client = Client(controller.hostname, controller.port)
>>> r = client.sendmail('a@example.com', ['b@example.com'], """\
... From: Anne Person <anne@example.com>
... To: Bart Person <bart@example.com>
... Subject: A test
... Message-ID: <ant>
...
... Hi Bart, this is Anne.
... """)
Message from a@example.com
Message for ['b@example.com']
Message data:
> From: Anne Person <anne@example.com>
> To: Bart Person <bart@example.com>
> Subject: A test
> Message-ID: <ant>
>
> Hi Bart, this is Anne.
End of message

You’ll notice that at the end of the DATA command, your handler’s handle_DATA() method was called. The sender, recipients, and message contents were taken from the envelope, and printed at the console. The handler methods also returns a successful status message.

ExampleHandler class also implements a handle_RCPT() method. This gets called after the RCPT TO command is sanity checked. The method ensures that all recipients are local to the @example.com domain, returning an error status if not. It is the handler’s responsibility to add valid recipients to the rcpt_tos attribute of the envelope and to return a successful status.

Thus, if we try to send a message to a recipient not inside example.com , it is rejected:

>>> client.sendmail('aperson@example.com', ['cperson@example.net'], """\
... From: Anne Person <anne@example.com>
... To: Chris Person <chris@example.net>
... Subject: Another test
... Message-ID: <another>
...
... Hi Chris, this is Anne.
... """)
Traceback (most recent call last):
...
smtplib.SMTPRecipientsRefused: {'cperson@example.net': (550, b'not relaying to that domain')}

When you’re done with the SMTP server, stop it via the controller.

>>> controller.stop()

The server is guaranteed to be stopped.

>>> client.connect(controller.hostname, controller.port)
Traceback (most recent call last):
...
ConnectionRefusedError: ...

There are a number of built-in handler classes that you can use to do some common tasks, and it’s easy to write your own handler. For a full overview of the methods that handler classes may implement, see the section on handler hooks .

Unix Socket-based Server ¶

UnixSocketController class creates a server listening to a Unix Socket (i.e., a special file that can act as a ‘pipe’ for interprocess communication).

Usage is identical with the example described in the TCP-based Server section above, with some differences:

Rather than specifying a hostname:port to listen on, you specify the Socket’s filepath:

>>> from aiosmtpd.controller import UnixSocketController
>>> from aiosmtpd.handlers import Sink
>>> controller = UnixSocketController(Sink(), unix_socket="smtp_socket~")
>>> controller.start()

Rather than connecting to IP:port, you connect to the Socket file. Python 的 smtplib.SMTP sadly cannot connect to a Unix Socket, so we need to handle it on our own here:

>>> import socket
>>> sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
>>> sock.connect("smtp_socket~")
>>> resp = sock.recv(1024)
>>> resp[0:4]
b'220 '

Try sending something, don’t forget to end with "\r\n" :

>>> sock.send(b"HELO example.org\r\n")
18
>>> resp = sock.recv(1024)
>>> resp[0:4]
b'250 '

And close everything when done:

>>> sock.send(b"QUIT\r\n")
6
>>> resp = sock.recv(1024)
>>> resp[0:4]
b'221 '
>>> sock.close()
>>> controller.stop()

启用 SMTPUTF8 ¶

It’s very common to want to enable the SMTPUTF8 ESMTP option, therefore this is the default for the Controller constructor. For backward compatibility reasons, this is not the default for the SMTP class though. If you want to disable this in the Controller , you can pass this argument into the constructor:

>>> from aiosmtpd.handlers import Sink
>>> controller = Controller(Sink(), enable_SMTPUTF8=False)
>>> controller.start()
>>>
>>> client = Client(controller.hostname, controller.port)
>>> code, message = client.ehlo('me')
>>> code
250

The EHLO response does not include the SMTPUTF8 ESMTP option.

>>> lines = message.decode('utf-8').splitlines()
>>> # Don't print the server host name line, since that's variable.
>>> for line in lines[1:]:
...     print(line)
SIZE 33554432
8BITMIME
HELP

Stop the controller if we’re done experimenting:

>>> controller.stop()

控制器 API ¶

class aiosmtpd.controller. IP6_IS ¶

NO : set ¶

YES : set ¶

aiosmtpd.controller. get_localhost ( ) ¶
返回
返回类型

class aiosmtpd.controller. BaseThreadedController ( handler , loop = None , * , ready_timeout , ssl_context = None , server_hostname = None , server_kwargs = None , ** SMTP_parameters ) ¶

handler

loop

ready_timeout : float ¶

server ¶

smtpd : aiosmtpd.smtp.SMTP ¶

方法

abstract _create_server ( ) → Coroutine ¶

abstract _trigger_server ( ) → None ¶

start ( ) → None ¶

stop ( ) → None ¶

Other parameters, attributes, and methods are identical to


BaseThreadedController

and thus are not repeated nor explained here.