如何使用 postmarker 编写 Celery 任务

Question

我正在尝试编写一个使用 postmarker client library 发送电子邮件的 celery 任务。这是一个调用 postmarkapp.com API 端点的简单客户端库。

# In tasks.py

@app.task(bind=True)
def send_email(...):
    """Sends a single email via Postmark."""
    postmark = PostmarkClient(...)
    response = postmark.emails.send(...)
    log.info('Postmark API send_email response %s', response)

但是当我尝试在测试中执行任务时：

# in test_tasks.py

class PostmarkSendEmail(TestCase):
    def test_call_send_email(self):
        send_email(...)

我收到以下错误

TypeError: <@task: tasks.send_email of groot at 0x7f2c35b9e911> is not JSON serializable

我的问题是，是否可以在 celery 任务中使用此客户端库？我可以做些什么来让我在 celery 任务中使用这个库吗？

Answer 1

我发现问题出在传递给 @app.task() 的 bind=True 参数上。解决方案是删除该参数：

@app.task()
def send_email(...):
    """Sends a single email via Postmark."""
    postmark = PostmarkClient(...)
    response = postmark.emails.send(...)
    log.info('Postmark API send_email response %s', response)

或者，我可以将 self 参数添加到 send_email 方法：

@app.task(bind=True)
def send_email(self, ...):
    """Sends a single email via Postmark."""
    postmark = PostmarkClient(...)
    response = postmark.emails.send(...)
    log.info('Postmark API send_email response %s', response)

作为bind参数，将方法绑定到App实例。

如何使用 postmarker 编写 Celery 任务

How to write a Celery task using postmarker

python

unit-testing

celery

django-celery

postmark