我一直在使用 Django 开发一个网络应用程序，我很好奇是否有办法安排作业定期运行。
基本上我只想通过数据库运行并自动定期进行一些计算/更新，但我似乎找不到任何关于这样做的文档。
有谁知道如何设置这个？
澄清一下：我知道我可以设置一个cron工作来做到这一点，但我很好奇 Django 中是否有一些特性可以提供这个功能。我希望人们能够自己部署这个应用程序，而不必做太多配置（最好是零）。
我已经考虑通过简单地检查自上次向站点发送请求以来是否应该运行作业来“追溯地”触发这些操作，但我希望有一些更清洁的东西。
我采用的一种解决方案是这样做：
1）创建自定义管理命令，例如
python manage.py my_cool_command
2）使用cron（在 Linux 上）或at（在 Windows 上）在需要的时间运行我的命令。
这是一个简单的解决方案，不需要安装繁重的 AMQP 堆栈。但是，使用其他答案中提到的芹菜之类的东西有很多好处。特别是，使用 Celery 时，不必将应用程序逻辑分散到 crontab 文件中是一件好事。但是，cron 解决方案非常适合中小型应用程序，并且您不希望有很多外部依赖项。
编辑：
在更高版本的 Windows 中at，Windows 8、Server 2012 及更高版本已弃用该命令。您可以schtasks.exe用于相同的用途。
**** 更新 **** 这是 django doc 的新链接，用于编写自定义管理命令
我们已经开源了我认为的结构化应用程序。布赖恩的上述解决方案也暗示了这一点。我们会喜欢任何/所有反馈！
https://github.com/tivix/django-cron
它带有一个管理命令：
./manage.py runcrons
这就是工作。每个 cron 都被建模为一个类（所以它都是面向对象的）并且每个 cron 以不同的频率运行，我们确保相同的 cron 类型不会并行运行（以防 cron 本身的运行时间比它们的频率长！）
有趣的新可插拔 Django 应用程序：django-chronograph
您只需添加一个用作计时器的 cron 条目，并且您可以在脚本中运行一个非常好的 Django 管理界面。
看看 Django Poor Man's 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
希望这对某人有帮助！
Brian Neal 建议通过 cron 运行管理命令效果很好，但如果您正在寻找更强大的东西（但不像 Celery 那样精致），我会研究像Kronos这样的库：
# app/cron.py
import kronos
@kronos.register('0 * * * *')
def task():
pass
用于调度程序作业的 Django APScheduler。Advanced Python Scheduler (APScheduler) 是一个 Python 库，可让您安排 Python 代码稍后执行，无论是一次还是定期执行。您可以随意添加新工作或删除旧工作。
注意：我是这个库的作者
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
RabbitMQ 和 Celery 比 Cron 有更多的特性和任务处理能力。如果任务失败不是问题，并且您认为您将在下一次调用中处理中断的任务，那么 Cron 就足够了。
Celery & AMQP会让你处理被破坏的任务，它会被另一个工作人员再次执行（芹菜工作人员监听下一个要处理的任务），直到max_retries达到任务的属性。您甚至可以在失败时调用任务，例如记录失败，或max_retries在达到目标后向管理员发送电子邮件。
当您需要扩展应用程序时，您可以分发 Celery 和 AMQP 服务器。
我个人使用 cron，但django-extensions的Jobs Scheduling部分看起来很有趣。
尽管不是 Django 的一部分，但Airflow是一个更新的项目（截至 2016 年），对任务管理很有用。
Airflow 是一个工作流自动化和调度系统，可用于创作和管理数据管道。基于 Web 的 UI 为开发人员提供了一系列用于管理和查看这些管道的选项。
Airflow 是用 Python 编写的，并使用 Flask 构建。
Airflow 由 Airbnb 的 Maxime Beauchemin 创建并于 2015 年春季开源。它于 2016 年冬季加入 Apache 软件基金会的孵化计划。这是Git 项目页面和一些附加背景信息。
将以下内容放在 cron.py 文件的顶部：
#!/usr/bin/python
import os, sys
sys.path.append('/path/to/') # the parent directory of the project
sys.path.append('/path/to/project') # these lines only needed if not on path
os.environ['DJANGO_SETTINGS_MODULE'] = 'myproj.settings'
# imports and code below
我只是想到了这个相当简单的解决方案：
您可以添加参数，但只需将参数添加到 URL。
告诉我你们的想法。
[更新]我现在使用django-extensions中的 runjob 命令而不是 curl。
我的 cron 看起来像这样：
@hourly python /path/to/project/manage.py runjobs hourly
... 以此类推，适用于每日、每月等。您还可以将其设置为运行特定作业。
我发现它更易于管理和更清洁。不需要将 URL 映射到视图。只需定义您的工作类别和 crontab 即可。
在部分代码之后，我可以写任何东西，就像我的views.py :)
#######################################
import os,sys
sys.path.append('/home/administrator/development/store')
os.environ['DJANGO_SETTINGS_MODULE']='store.settings'
from django.core.management impor setup_environ
from store import settings
setup_environ(settings)
#######################################
来自 http://www.cotellese.net/2007/09/27/running-external-scripts-against-django-models/
你一定要看看 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,
},
}
一个更现代的解决方案（与 Celery 相比）是 Django Q： https ://django-q.readthedocs.io/en/latest/index.html
它有很好的文档并且很容易理解。缺少 Windows 支持，因为 Windows 不支持进程分叉。但是，如果您使用 Windows for Linux 子系统创建开发环境，它就可以正常工作。
我今天遇到了类似你的问题。
我不想让服务器通过 cron 处理它（大多数库最终只是 cron 助手）。
所以我创建了一个调度模块并将其附加到init。
这不是最好的方法，但它可以帮助我将所有代码放在一个地方，并且它的执行与主应用程序相关。
是的，上面的方法太棒了。我尝试了其中的一些。最后，我找到了这样的方法：
from threading import Timer
def sync():
do something...
sync_timer = Timer(self.interval, sync, ())
sync_timer.start()
就像递归一样。
好的，我希望这个方法能满足你的要求。:)
我使用 celery 来创建我的定期任务。首先，您需要按如下方式安装它：
pip install django-celery
不要忘记django-celery在您的设置中注册，然后您可以执行以下操作：
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
我不确定这对任何人是否有用，因为我必须提供系统的其他用户来安排作业，而没有让他们访问实际的服务器（Windows）任务计划程序，我创建了这个可重用的应用程序。
请注意，用户可以访问服务器上的一个共享文件夹，他们可以在其中创建所需的命令/任务/.bat 文件。然后可以使用此应用程序安排此任务。
截屏：
如果您想要比Celery更可靠的东西，请尝试构建在AWS SQS/SNS之上的TaskHawk。
对于简单的 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
models/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',
...
]