我一直在使用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',
...
]
其他回答
我今天遇到了类似的问题。
我不想让服务器通过cron来处理它(而且大多数库最终只是cron助手)。
所以我已经创建了一个调度模块，并将其附加到init。
这不是最好的方法，但它帮助我把所有的代码放在一个地方，它的执行与主应用程序相关。
是的，上面的方法太棒了。我尝试了其中一些。最后，我找到了这样一个方法:
from threading import Timer
def sync():
do something...
sync_timer = Timer(self.interval, sync, ())
sync_timer.start()
就像递归一样。
好的，我希望这个方法能满足你的要求。：）
Django APScheduler调度器。Advanced Python Scheduler (APScheduler)是一个Python库，可以安排Python代码稍后执行，可以只执行一次，也可以定期执行。您可以随时添加新作业或删除旧作业。
注:我是这个库的作者
安装APScheduler
pip install apscheduler
查看要调用的文件函数
文件名称:scheduler_jobs.py
def FirstCronTest():
print("")
print("I am executed..!")
配置调度程序
创建execute.py文件并添加以下代码
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
在这里，调度器函数写在scheduler_jobs中
import scheduler_jobs
scheduler.add_job(scheduler_jobs.FirstCronTest, 'interval', seconds=10)
scheduler.start()
链接文件以执行
现在，在Url文件的底部添加下面的行
import execute
您可以通过执行检查完整的代码 (点击这里) https://github.com/devchandansh/django-apscheduler
我个人使用cron，但是django扩展的作业调度部分看起来很有趣。
我有完全相同的需求一段时间前，并最终解决它使用APScheduler(用户指南)
它使调度任务超级简单，并使其独立于某些代码的基于请求的执行。下面是一个简单的例子。
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
job = None
def tick():
print('One tick!')\
def start_job():
global job
job = scheduler.add_job(tick, 'interval', seconds=3600)
try:
scheduler.start()
except:
pass
希望这能帮助到一些人!
