mailqueue-runner 0.12.1

mailqueue-runner

此库提供了一种可靠的方式来将电子邮件消息发送到外部SMTP服务器。该API被设计成易于集成到您的Python（Web）应用中。

此外，还有一些CLI脚本（功能有限）允许应用程序通过CLI发送电子邮件，就像传统上使用/usr/bin/sendmail和/usr/bin/mail一样。因此，对于非常简单的用例，此软件也是msmtp和ssmtp的替代品。

核心代码包含一个排队系统来处理发送消息时的（临时）错误（例如，中断的连接），并提供详细的错误日志。

当消息无法通过SMTP发送时，它可以存储在磁盘上的类似maildir的队列中。外部辅助脚本（mq-run）在稍后时间取回它们并尝试再次投递这些消息。辅助脚本必须定期调用（例如，通过cron）。

作为一项额外的优点，该库非常模块化，因此您可以将自定义代码插入其中，并根据需要调整库。

安装

$ pip install mailqueue-runner

如果您不感兴趣Python集成，只想使用mq-sendmail，您可以使用我的COPR仓库为Fedora和RHEL

$ dnf copr enable fschwarz/mailqueue-runner
$ dnf install mailqueue-runner

使用mq-sendmail（CLI）

代码提供了一个名为mq-sendmail的CLI应用程序，它提供了与常见的Un*x /usr/bin/sendmail应用程序的（基本）兼容性。此外，它还支持msmtp添加的一些方便参数。

$ mq-sendmail --set-date-header --set-msgid-header root <<<MAIL
Subject: Test email
From: me@site.example
MIME-Version: 1.0
Content-Transfer-Encoding: 8bit
Content-Type: text/plain; charset=UTF-8

mail body
MAIL

默认情况下，配置是从~/.mailqueue-runner.conf或/etc/mailqueue-runner.conf读取的，尽管您也可以使用--config=...显式指定配置文件。与其他sendmail实现类似，应用程序解析/etc/aliases以查找收件人的电子邮件地址。

请注意，只有当配置文件包含queue_dir选项时，代码才会将消息排队在失败投递之后。

使用mq-mail（CLI）

该代码提供了一个名为 mq-mail 的命令行应用程序，该程序与 /usr/bin/mail 应用程序具有（基本）兼容性。

$ mq-mail --from-address="me@site.example" --subject "subject" root <<<MAIL
mail body
MAIL

默认情况下，配置从 ~/.mailqueue-runner.conf 或 /etc/mailqueue-runner.conf 读取，尽管您也可以使用 --config=... 显式指定配置文件。应用程序解析 /etc/aliases 来查找收件人的电子邮件地址。

请注意，只有当配置文件包含queue_dir选项时，代码才会将消息排队在失败投递之后。

配置（命令行脚本）

配置文件使用传统的“ini”样格式

[mqrunner]
smtp_hostname = hostname
smtp_port = 587
smtp_username = someuser@site.example
smtp_password = secret
# optional but the CLI scripts will not queue messages if this is not set
queue_dir = /path/to/mailqueue
# optional, SMTP envelope from (also used when "--set-from-header" is given)
from = user@host.example
# optional, format as described in
# https://docs.pythonlang.cn/3/library/logging.config.html#logging-config-fileformat
# logging_conf = /path/to/logging.conf
# optional, ignored if "logging_conf" is set
delivery_log = /path/to/delivery.log
# optional, ignored if "logging_conf" is set
queue_log = /path/to/queue.log

有关关于包装 mq-run（例如，重用现有配置格式）的更多信息，请参阅Cookbook：mq-run 的自定义包装。

用法（邮件提交/Python 集成）

from schwarz.mailqueue import init_smtp_mailer, MaildirBackend, MessageHandler
# settings: a dict-like instance with keys as shown below in the "Configuration" section
settings = {}
# Adapt the list of transports as you like (ordering matters):
# - always enqueue: use "MaildirBackend()" only
# - never enqueue: use "init_smtp_mailer()" only
transports = [
    init_smtp_mailer(settings),
    MaildirBackend('/path/to/queue-dir'),
]
handler = MessageHandler(transports)
msg = b'…' # RFC-822/RFC-5322 message as bytes or email.Message instance
was_sent = handler.send_message(msg, sender='foo@site.example', recipient='bar@site.example')
# "was_sent" evaluates to True if the message was sent via SMTP or queued
# for later delivery.
was_queued = (getattr(send_result, 'queued', None) is not False)

用法（mq-run）

如果您使用队列处理临时投递问题，则需要定期运行脚本以重试投递。mq-run 提供此功能

$ mq-run

如果您想测试您的配置，可以向发送测试消息以确保邮件流设置正确

$ mq-send-test --to=recipient@site.example

日志记录

日志可以帮助您监控邮件处理。根据投递类型，库使用两个不同的记录器

• mailqueue.delivery_log：消息已投递到 SMTP 服务器
• mailqueue.queue_log：消息已排队，将由 mq-run 在以后投递

对于 CLI 应用程序（例如 mq-run、mq-sendmail、mq-mail），您可以在配置文件中设置 delivery_log = /path/to/delivery.log 和 queue_log = ... 以存储日志消息。如果设置了完全自定义的日志配置（logging_config = ...）或使用 mailqueue-runner 作为纯 Python 库，则忽略这些配置选项。

插件

该库允许通过插件自定义消息处理。插件使用 Puzzle 插件系统（blinker+setuptools）构建。插件支持是可选的，并需要额外的 PuzzlePluginSystem 依赖项（pip install mailqueue-runner[plugins]）。

插件可以实现的特性

• 有关成功/失败投递的通知（例如，额外的日志记录，将某些数据存储在外部数据库中等）
• 在失败投递尝试后丢弃排队消息（例如，在 10 次失败尝试后放弃）

有关插件发现/插件开发的更多信息，请参阅 Puzzle 插件项目。

CLI 工具（如 mq-run）会加载您的插件，如果将其添加到扩展点 mailqueue.plugins，例如

# setup.cfg (of your custom app)
[options.entry_points]
mailqueue.plugins =
    myplugin = example.app.mqplugin

示例插件代码

# example/app/mqplugin.py
from schwarz.puzzle_plugins import connect_signals, disconnect_signals
from schwarz.mailqueue import registry, MQAction, MQSignal

class MyPlugin:
    def __init__(self, registry):
        self._connected_signals = None
        self._registry = registry

    def signal_map(self):
        return {
            MQSignal.delivery_successful: self.delivery_successful,
            MQSignal.delivery_failed: self.delivery_failed,
        }

    def delivery_successful(self, _, msg, send_result):
        # called when a message was delivered successfully
        pass

    def delivery_failed(self, _, msg, send_result):
        # called when message delivery failed
        if msg.retries > 10:
            # discard messsage after 10 failed delivery attempts
            return MQAction.DISCARD
        return None

def initialize(context, registry):
    plugin = MyPlugin(registry)
    plugin._connected_signals = connect_signals(plugin.signal_map(), registry)
    context['plugin'] = plugin

def terminate(context):
    plugin = context['plugin']
    disconnect_signals(plugin._connected_signals, plugin._registry)
    plugin._registry = None
    plugin._connected_signals = None

Cookbook：mq-run 的自定义包装

虽然 mq-run 通常表现良好，但有时您可能希望有更多的控制。例如，您可能不想重复配置（一次用于您的实际应用程序，一次用于 mq-run）。好消息是您可以编写一个非常简单的包装器，以利用现有的代码而不重复 mq-run 的功能

#!/usr/bin/env python3

from schwarz.mailqueue.queue_runner import one_shot_queue_run

def main():
    # set up custom configuration, logging here (use your existing code)
    cli_options = {'verbose': True}
    # prepare configuration as expected by mailqueue-runner
    settings = {
        # … (smtp settings)

        # --- optional ---
        # only load "myplugin" plugin
        'plugins': 'myplugin',
        # do not reset currently configured loggers, just add a few for UI output
        'basic_logging_configured': True,
        # ability to inject a custom MessageHandler instance for maximum flexibility
        #'mh': …
    }
    one_shot_queue_run(queue_dir, options=cli_options, settings=settings)

if __name__ == '__main__':
    main()

Cookbook：保守的消息发送

上述默认配置尝试通过 SMTP 发送消息（如果可能），并且仅在 SMTP 投递失败时将数据序列化到持久存储（文件系统）。这种方法通常是性能（将数据序列化到磁盘较慢）和确保消息最终会发送之间的良好折衷。

但是，有时确保您不会丢失任何消息（即使 mailqueue-runner 有错误并且直接在尝试通过 SMTP 发送消息后崩溃）非常重要。为了降低此风险，您可以在尝试通过 SMTP 发送消息之前将消息持久化存储。

from schwarz.mailqueue import enqueue_message, MessageHandler

md_msg = enqueue_message(msg, path_maildir,
    sender      = '...',
    recipients  = ('...',),
    in_progress = True,
    return_msg  = True,
)
handler = MessageHandler(transports=...)
was_sent = handler.send_message(md_msg, sender='foo@site.example', recipient='bar@site.example')

请注意，您不需要在应用程序中仅使用单一的方法。对于非常重要的消息，您可以像上面显示的那样使用保守的消息发送，而对于大多数不那么重要的消息，则依赖于以性能为导向的方法。

动机/相关软件

许多Web应用程序需要发送电子邮件。通常，这是通过将消息发送到真实的SMTP服务器来完成的，然后该服务器将消息分发到网络上的远程邮件服务器（好吧，现在大多是Gmail）。直到您的SMTP服务器不可达（例如，网络错误）或由于暂时性错误（例如，DNS故障，无法验证发件人）不接受消息，一切才会顺利。

"mailqueue-runner"实现了一个（持久）消息队列，并提供了一个帮助可靠发送电子邮件的脚本（假设您有足够的空闲磁盘空间）。

repoze.sendmail 类似且是一款可靠的软件。我写了另一个库，因为我想要

• 避免SMTP服务器由于（暂时性）错误不接受消息而造成数据丢失，同时在不影响大部分时间（即大多数时间）正常工作的情况下延迟消息
• 避免SMTP服务器在多收件人消息中拒绝一个（但不是所有）收件人而带来的意外情况
• 不同的错误处理/更好的集成到自定义Web应用程序（投递日志，错误处理）
• 更好的错误日志记录（包括记录完整的SMTP对话的能力）
• 对队列中的消息进行最小修改（repoze.sendmail使用Python的email模块来操作投递所需的邮件头）

django-mail-queue 提供了一个Django应用程序，也提供了一个Web界面。显然，它仅限于Django，并且有太多的隐式（“魔法”）行为，不符合我的口味。然而，如果您使用Django，我推荐尝试这个库。

非目标

• 没有实际生成电子邮件的代码（例如，从模板生成，添加附件等）
• 当SMTP服务器不可用时，可能不适合高容量消息发送（> 100条消息/秒），因为消息将存储在（慢速的）文件系统上。

测试的Python版本

项目使用GitHub Actions运行测试套件。希望这意味着所有测试版本都适用于生产。目前，我测试了Python 3.6-3.12以及Linux上的pypy3。不推荐在Windows上部署，因为所有锁定都将禁用（由于Windows无法删除/移动打开的文件），但在Windows机器上进行开发应该是没有问题的。

许可证

代码采用MIT许可证，只有少数例外：它包含Python的smtplib的定制（略作修改）版本，该版本采用Python License 2.0。