我一直在使用Django开发一个web应用程序，我很好奇是否有一种方法可以安排一个作业定期运行。
基本上，我只是想运行数据库，并在自动的、定期的基础上进行一些计算/更新，但我似乎找不到任何关于这样做的文档。
有人知道怎么设置吗?
澄清一下:我知道我可以设置一个cron作业来完成这个任务，但我很好奇Django中是否有一些特性提供了这个功能。我希望人们能够自己部署这个应用程序，而不需要做很多配置(最好是零配置)。
我曾经考虑过“回溯性”地触发这些操作，方法是简单地检查自上一次请求发送到站点以来作业是否应该运行，但我希望使用更简洁的方法。
当前回答
另一种选择是使用火箭技术:
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本质上只是一个复杂的循环，可以生成异步任务，线程和进程。
声明:我是作者。
其他回答
芹菜是一个分布式任务队列，建立在AMQP (RabbitMQ)上。它还以类似cron的方式处理周期性任务(参见周期性任务)。根据你的应用，它可能值得一看。
用django (docs)设置芹菜非常容易，周期性任务实际上会在停机的情况下跳过错过的任务。芹菜还有内置的重试机制，以防任务失败。
RabbitMQ和芹菜比Cron有更多的特性和任务处理能力。如果任务失败不是问题，并且您认为将在下一个调用中处理中断的任务，那么Cron就足够了。
Celery & AMQP将允许您处理中断的任务，它将由另一个worker再次执行(Celery worker侦听下一个要处理的任务)，直到到达任务的max_retries属性。您甚至可以在失败时调用任务，比如记录失败，或者在到达max_retries时向管理员发送电子邮件。
当需要扩展应用程序时，可以分发芹菜和AMQP服务器。
你一定要看看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,
},
}
是的，上面的方法太棒了。我尝试了其中一些。最后，我找到了这样一个方法:
from threading import Timer
def sync():
do something...
sync_timer = Timer(self.interval, sync, ())
sync_timer.start()
就像递归一样。
好的，我希望这个方法能满足你的要求。：）
如果你想要比芹菜更可靠的东西，可以尝试构建在AWS SQS/SNS之上的TaskHawk。
参见:http://taskhawk.readthedocs.io
