放假的电话

Celery with test database: matching query does not exist

In a recent project, I used Django with Celery. When testing, I ran into some problem and it took me a while to solve. Hope this can help you a bit if you see the same thing.

In my project, I tried to use Celery to send email. Meanwhile, I have some model called EmailLog to track the emails. Part of the code is like this:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@shared_task
def _send_email(subject, to, email_log_pk):
email = EmailMessage()
email_log = EmailLog.objects.get(pk=email_log_pk)
try:
email.send()
email_log.status = 'Sent'
except:
email_log.status = 'Failed'
finally:
email_log.save()
def send_my_email(subject, to):
email_log = EmailLog(status='Pending')
email_log.save()
print email_log
_send_email.delay(subject, to, email_log.pk)
print email_log

In my send_my_email function, I created an EmailLog object, saved it first and then send the email using Celery. In the Celery task, the email_log’s status will be updated. The two print statements ensure that email_log is actually in the database.

While running unittest, however, my Celery worker can’t find the saved EmailLog object, saying ‘matching query does not exist’. It seems that the tests are wrapped in some transaction and even if both Django and Celery are configured to use the same database, they can’t share data.

Finally, I set CELERY_ALWAYS_EAGER=True to bypass this problem. It will execute the task locally, not in the queue. At least it can verify the correctness of the function.