为本地开发人员设置 django-celery-email

Question

似乎从django-allauth应用程序异步发送电子邮件的最佳方法是简单地安装django-celery-email。 但包裹警告说，

此版本需要以下版本：Python 2.7 和 Python3.5、Django 1.11、2.1 和 2.2 Celery 4.0

我只使用python几个月，从未遇到过一个项目需要两个python版本的情况。 我正在使用 pipenv 的官方推荐进行本地开发。 快速谷歌显示，不可能在虚拟环境中安装两个python解释器。 由于该插件看起来如此受欢迎，我想知道其他人是如何设置它的？ 如果我错过了一些解释这一点的重大内容，我们深表歉意。

一个额外的答案还会考虑到我正在使用 docker，并且 docker 映像将像这样安装 python 包。

RUN pipenv install --system --deploy --ignore-pipfile

提前非常感谢。

Answer 1

我很确定这只是项目文档中的不准确描述，所以你需要安装 python 2.7 或 python>=3.5

Answer 2

最后我没有使用django-celery-email。 无需此软件包即可轻松发送由 django-allauth 应用程序生成的电子邮件。

我使用了这些资源 -

https://github.com/anymail/django-anymail/issues/79

https://docs.djangoproject.com/en/2.2/topics/email/#defining-a-custom-email-backend

基本上你这样做是为了让它工作。

在 settings.py 中定义自定义电子邮件后端 -

EMAIL_BACKEND = "users.backends.CustomEmailBackend"

在 backend.py 文件中定义后端 -

from django.core.mail.backends.base import BaseEmailBackend
from .tasks import async_send_messages
class CustomEmailBackend(BaseEmailBackend):
def send_messages(self, email_messages):
async_send_messages.delay(email_messages)
return len(email_messages)

这就是任务——

from django.core.mail import get_connection
from abstract_base_user.celery import app
@app.task(rety_backoff=True, serializer="pickle")
def async_send_messages(email_messages):
conn = get_connection(backend='anymail.backends.mailgun.EmailBackend')
conn.send_messages(email_messages)

芹菜 django 应用程序应按照 https://docs.celeryproject.org/en/latest/django/first-steps-with-django.html 中定义的标准方式进行设置

而 settings.py 需要的芹菜设置包括泡菜内容类型——

CELERY_ACCEPT_CONTENT = ['json', 'pickle']

显然，您还需要包括您的随时设置和经纪人设置。 但这应该足以让任何人开始。

Answer 3

我决定让它变得更好。用泡菜序列化是不可接受的 imo。

class AsyncSmtpEmailBackend(SmtpEmailBackend):

def serialize_message(self, email_message):
if not email_message.recipients():
return False
encoding = email_message.encoding or settings.DEFAULT_CHARSET
from_email = sanitize_address(email_message.from_email, encoding)
recipients = [
sanitize_address(addr, encoding) for addr in email_message.recipients()
]
msg = email_message.message()
msg_data = msg.as_bytes(linesep="rn")
charset = (
msg.get_charset().get_output_charset() if msg.get_charset() else "utf-8"
)
msg_data = msg_data.decode(charset)
serialized_data = {
"from_email": from_email,
"recipients": recipients,
"message": msg_data,
}
return json.dumps(serialized_data)

def send_messages(self, email_messages):
msgs = [self.serialize_message(msg) for msg in email_messages]
async_send_messages_with_smtp.delay(msgs)
return len(email_messages)

def _send(self, email_message):
email_message = json.loads(email_message)
from_email = email_message['from_email']
recipients = email_message['recipients']
message = email_message['message']
try:
self.connection.sendmail(
from_email, recipients, message
)
except smtplib.SMTPException:
if not self.fail_silently:
raise
return False
return True

并在tasks.py

@shared_task()
def async_send_messages_with_smtp(email_messages):
conn = get_connection(backend=settings.EMAIL_BACKEND)
if not email_messages:
return 0
with conn._lock:
new_conn_created = conn.open()
if not conn.connection or new_conn_created is None:
# We failed silently on open().
# Trying to send would be pointless.
return 0
num_sent = 0
for message in email_messages:
sent = conn._send(message)
if sent:
num_sent += 1
if new_conn_created:
conn.close()
return num_sent