如何使用Celery和Docker处理Django中的定期任务

class Command(BaseCommand):
help = "A description of the command"

def handle(self, *args, **options):
self.stdout.write("My sample command just ran.") # NEW

测试：

$ docker-compose exec web python manage.py my_custom_command
My sample command just ran.

这样，让我们将所有内容捆绑在一起！

使用Celery Beat安排自定义命令

现在我们已经启动并运行了容器，已经过测试，可以安排任务定期运行，并编写了自定义的Django Admin示例命令，现在该进行设置以定期运行自定义命令了。

设定

在项目中，我们有一个非常基本的应用程序，称为订单。 它包含两个模型，产品和订单。 让我们创建一个自定义命令，该命令从当天发送确认订单的电子邮件报告。

首先，我们将通过此项目中包含的夹具将一些产品和订单添加到数据库中：

$ docker-compose exec web python manage.py loaddata products.json

创建超级用户：

$ docker-compose exec web python manage.py createsuperuser

出现提示时，请填写用户名，电子邮件和密码。 然后在您的Web浏览器中导航到http://127.0.0.1:1337/admin。 使用您刚创建的超级用户登录，并创建几个订单。 确保至少有一个日期为今天。

让我们为我们的电子邮件报告创建一个新的自定义命令。

创建一个名为orders / management / commands / email_report.py的文件：

from datetime import timedelta, time, datetime
from django.core.mail import mail_admins
from django.core.management import BaseCommand
from django.utils import timezone
from django.utils.timezone import make_aware
from orders.models import Order
today = timezone.now()
tomorrow = today + timedelta(1)
today_start = make_aware(datetime.combine(today, time()))
today_end = make_aware(datetime.combine(tomorrow, time()))
class Command(BaseCommand):
 help = "Send Today's Orders Report to Admins"
 def handle(self, *args, **options):
 orders = Order.objects.filter(confirmed_date__range=(today_start, today_end))
 if orders:
 message = ""
 for order in orders:
 message += f"{order} n"
 subject = (
 f"Order Report for {today_start.strftime('%Y-%m-%d')} "
 f"to {today_end.strftime('%Y-%m-%d')}"
 )
 mail_admins(subject=subject, message=message, html_message=None)
 self.stdout.write("E-mail Report was sent.")
 else:
 self.stdout.write("No orders confirmed today.")

在代码中，我们向数据库查询了日期为Confirmed_date的订单，将订单合并为电子邮件正文的单个消息，然后使用Django内置的mail_admins命令将电子邮件发送给管理员。