我一直在使用Django开发一个web应用程序，我很好奇是否有一种方法可以安排一个作业定期运行。
基本上，我只是想运行数据库，并在自动的、定期的基础上进行一些计算/更新，但我似乎找不到任何关于这样做的文档。
有人知道怎么设置吗?
澄清一下:我知道我可以设置一个cron作业来完成这个任务，但我很好奇Django中是否有一些特性提供了这个功能。我希望人们能够自己部署这个应用程序，而不需要做很多配置(最好是零配置)。
我曾经考虑过“回溯性”地触发这些操作，方法是简单地检查自上一次请求发送到站点以来作业是否应该运行，但我希望使用更简洁的方法。
当前回答
对于简单的dockerized项目，我真的看不到任何现有的答案适合。
所以我写了一个非常简单的解决方案，不需要外部库或触发器，它可以自己运行。不需要外部os-cron，应该可以在任何环境下工作。
它通过添加一个中间件来工作:middleware.py
import threading
def should_run(name, seconds_interval):
from application.models import CronJob
from django.utils.timezone import now
try:
c = CronJob.objects.get(name=name)
except CronJob.DoesNotExist:
CronJob(name=name, last_ran=now()).save()
return True
if (now() - c.last_ran).total_seconds() >= seconds_interval:
c.last_ran = now()
c.save()
return True
return False
class CronTask:
def __init__(self, name, seconds_interval, function):
self.name = name
self.seconds_interval = seconds_interval
self.function = function
def cron_worker(*_):
if not should_run("main", 60):
return
# customize this part:
from application.models import Event
tasks = [
CronTask("events", 60 * 30, Event.clean_stale_objects),
# ...
]
for task in tasks:
if should_run(task.name, task.seconds_interval):
task.function()
def cron_middleware(get_response):
def middleware(request):
response = get_response(request)
threading.Thread(target=cron_worker).start()
return response
return middleware
模型/ cron.py:
from django.db import models
class CronJob(models.Model):
name = models.CharField(max_length=10, primary_key=True)
last_ran = models.DateTimeField()
settings.py:
MIDDLEWARE = [
...
'application.middleware.cron_middleware',
...
]
其他回答
我只是想到了一个相当简单的解决方案:
定义一个视图函数do_work(req, param)，就像使用其他任何视图一样，使用URL映射，返回一个HttpResponse等等。 使用您的定时首选项(或在Windows中使用AT或Scheduled Tasks)设置一个运行curl http://localhost/your/mapped/url?param=value的cron作业。
你可以添加参数，但只是将参数添加到URL。
告诉我你们的想法。
[更新]我现在使用runjob命令从django扩展而不是curl。
我的cron看起来是这样的:
@hourly python /path/to/project/manage.py runjobs hourly
．.． 每天，每月等等。还可以将其设置为运行特定作业。
我觉得这样更容易管理，也更干净。不需要将URL映射到视图。只需定义作业类和crontab，就完成了。
我用芹菜来创建我的定期任务。首先你需要安装它，如下所示:
pip install django-celery
不要忘记在你的设置中注册django-芹菜，然后你可以这样做:
from celery import task
from celery.decorators import periodic_task
from celery.task.schedules import crontab
from celery.utils.log import get_task_logger
@periodic_task(run_every=crontab(minute="0", hour="23"))
def do_every_midnight():
#your code
另一种选择是使用火箭技术:
from rocketry import Rocketry
from rocketry.conds import daily, after_success
app = Rocketry()
@app.task(daily.at("10:00"))
def do_daily():
...
@app.task(after_success(do_daily))
def do_after_another():
...
if __name__ == "__main__":
app.run()
它还支持自定义条件:
from pathlib import Path
@app.cond()
def file_exists(file):
return Path(file).exists()
@app.task(daily & file_exists("myfile.csv"))
def do_custom():
...
它也支持Cron:
from rocketry.conds import cron
@app.task(cron('*/2 12-18 * Oct Fri'))
def do_cron():
...
它可以很好地与FastAPI集成，我认为它也可以与Django集成，Rocketry本质上只是一个复杂的循环，可以生成异步任务，线程和进程。
声明:我是作者。
你一定要看看django-q! 它不需要额外的配置，并且很可能具备在商业项目中处理任何生产问题所需的一切。
它是积极开发的，与django, django ORM, mongo, redis集成得很好。以下是我的配置:
# django-q
# -------------------------------------------------------------------------
# See: http://django-q.readthedocs.io/en/latest/configure.html
Q_CLUSTER = {
# Match recommended settings from docs.
'name': 'DjangoORM',
'workers': 4,
'queue_limit': 50,
'bulk': 10,
'orm': 'default',
# Custom Settings
# ---------------
# Limit the amount of successful tasks saved to Django.
'save_limit': 10000,
# See https://github.com/Koed00/django-q/issues/110.
'catch_up': False,
# Number of seconds a worker can spend on a task before it's terminated.
'timeout': 60 * 5,
# Number of seconds a broker will wait for a cluster to finish a task before presenting it again. This needs to be
# longer than `timeout`, otherwise the same task will be processed multiple times.
'retry': 60 * 6,
# Whether to force all async() calls to be run with sync=True (making them synchronous).
'sync': False,
# Redirect worker exceptions directly to Sentry error reporter.
'error_reporter': {
'sentry': RAVEN_CONFIG,
},
}
一个更现代的解决方案(与芹菜相比)是Django Q: https://django-q.readthedocs.io/en/latest/index.html
它有很好的文档，很容易理解。缺乏Windows支持，因为Windows不支持进程分叉。但是如果您使用Windows for Linux子系统创建您的开发环境，那么它工作得很好。
