python - 为 `ClassTask` mixin 子类化 Celery Task

Question

以我是 Celery 新手这一事实作为我的问题的开头，这(1)可能已经在其他地方得到了回答(如果是这样，我找不到答案)或(2)可能有更好的方法实现我的目标而不是我直接要求的目标。

此外，我知道 celery.contrib.methods，但 task_method 并不能完全实现我正在寻找的目标。

我的目标

我想创建一个类 mixin，将整个类转换为 Celery 任务。例如，一个 mixin 由类似下面的代码表示(现在无法运行):

from celery import Task

class ClassTaskMixin(Task):

    @classmethod
    def enqueue(cls, *args, **kwargs):
        cls.delay(*args, **kwargs)

    def run(self, *args, **kwargs):
        Obj = type(self.name, (), {})
        Obj(*args, **kwargs).run_now()

    def run_now(self):
        raise NotImplementedError()

与使用 task_method 时不同，我不想在任务排队和调用 .delay() 之前完全实例化该类。相反，我想简单地将类名称以及任何相关的初始化参数传递给异步进程。然后，异步进程将使用类名和给定的初始化参数完全实例化该类，然后在实例化的对象上调用某些方法(例如 .run_now())。

示例用例

异步构建和发送电子邮件将是我需要的 mixin 的一个示例。

class WelcomeEmail(EmailBase, ClassTaskMixin):

    def __init__(self, recipient_address, template_name, template_context):
        self.recipient_address = recipient_address
        self.template_name = template_name
        self.template_context = template_context

   def send(self):
       self.render_templates()
       self.construct_mime()
       self.archive_to_db()
       self.send_smtp_email()

   def run_now(self):
       self.send()

上面的代码将通过调用 WelcomeEmail.enqueue(recipient_address, template_name, template_context) 在异步 Celery 进程中发送电子邮件。通过调用 WelcomeEmail(recipient_address, template_name, template_context).send() 可以在进程中同步发送电子邮件。

问题

1. 是否有任何理由表明我尝试做的事情在 Celery 框架内是非常非常错误的？
2. 是否有更好的方法来构建 mixin，使其比我所建议的更加 Celery-onic(更好的属性名称、不同的方法结构等)？
3. 要使 mixin 在我所描述的用例中发挥作用，我缺少什么？

Answer 1

显然这个问题对很多人来说并不是很有趣，但是......我已经完成了我打算做的事情。

请参阅拉取请求 https://github.com/celery/celery/pull/1897了解详情。