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

rmbxnbpk  于 2023-08-08  发布在  Go
关注(0)|答案(3)|浏览(102)

似乎从django-allauth应用程序异步发送电子邮件的最佳方法是简单地安装django-celery-email。但 Package 上警告说
此版本需要以下版本:Python 2.7和Python 3.5,Django 1.11,2.1和2.2 Celery 4.0
我只使用了几个月的python,从来没有遇到过一个项目需要两个python版本的情况。我正在使用pipenv的官方推荐用于当地发展。快速google显示,在虚拟环境中不可能安装两个python解释器。由于插件似乎如此受欢迎,我想知道其他人是如何设置它的?如果我错过了一些重要的东西来解释这一点,我道歉。
一个额外的答案还将考虑到我正在使用docker,docker镜像将安装像这样的python包。
第一个月
提前感谢。

92dk7w1h

92dk7w1h1#

我很确定这只是project docs中的不准确描述,所以您需要安装python 2.7或python >=3.5。

a5g8bdjr

a5g8bdjr2#

最后我没有使用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中定义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)


celery django应用程序应该按照https://docs.celeryproject.org/en/latest/django/first-steps-with-django.html中定义的标准方式进行设置。
并且settings.py中的celery设置需要包括pickle内容类型-
CELERY_ACCEPT_CONTENT = ['json', 'pickle']
显然,您需要包括您的随时设置和经纪人设置太多。但这足以让任何人开始。

whitzsjs

whitzsjs3#

我决定让这一切变得更好。用pickle进行序列化在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="\r\n")
        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

相关问题