设置预定的工作？

Question

我一直在使用 Django 开发一个网络应用程序，我很好奇是否有一种方法可以安排作业定期运行。

基本上我只想运行数据库并自动定期进行一些计算/更新，但我似乎找不到任何有关执行此操作的文档。

有谁知道如何设置这个？

澄清一下：我知道我可以设置一个 cron 作业来执行此操作，但我很好奇 Django 中是否有某些功能可以提供此功能。 我希望人们能够自己部署这个应用程序，而无需进行太多配置（最好是零）。

我考虑过通过简单地检查自上次将请求发送到站点以来是否应该运行作业来“追溯”触发这些操作，但我希望有一些更干净的东西。

Answer 1

我采用的一种解决方案是执行此操作：

1）创建 自定义管理命令，例如

python manage.py my_cool_command

2) 使用 cron（在 Linux 上）或 at（在 Windows 上）在所需次数。

这是一个简单的解决方案，不需要安装繁重的 AMQP 堆栈。 然而，使用其他答案中提到的像 Celery 这样的东西有很好的优势。 特别是，使用 Celery 时，不必将应用程序逻辑分散到 crontab 文件中，这是件好事。 然而，cron 解决方案对于中小型应用程序以及您不需要大量外部依赖项的应用程序非常有效。

编辑：

在更高版本的 Windows 中，Windows 8、Server 2012 及更高版本已弃用 at 命令。 您可以使用 schtasks.exe 来实现相同的用途。

**** 更新 ****
这是新的链接用于编写自定义管理命令的 django doc

Answer 2

Celery 是一个分布式任务队列，基于 AMQP (RabbitMQ) 构建。 它还以类似 cron 的方式处理定期任务（请参阅定期任务）。 根据您的应用程序，它可能值得一看。

Celery 使用 django 非常容易设置（docs），并且定期任务实际上会在停机时跳过错过的任务。 Celery 还具有内置的重试机制，以防任务失败。

Answer 3

我们开源了我认为是结构化应用程序的源代码。 布莱恩的解决方案上面也提到了。 我们希望得到任何/所有反馈！

https://github.com/tivix/django-cron

带有一个管理命令：

./manage.py runcrons

它 完成工作。 每个 cron 都被建模为一个类（因此都是面向对象的），并且每个 cron 以不同的频率运行，我们确保相同的 cron 类型不会并行运行（以防 cron 本身运行的时间比它们的频率更长！）

Answer 4

如果您使用标准 POSIX 操作系统，则使用 cron。

如果您使用的是 Windows，请使用 at。

编写 Django 管理命令来

1. 找出它们所在的平台。

2. 为您的用户执行适当的“AT”命令，或为您的用户更新 crontab。

   为您的用户

Answer 5

有趣的新可插拔 Django 应用程序： django-chronograph

您只需添加一个 cron 条目即可作为一个计时器，你有一个非常好的 Django 管理界面来运行脚本。

Answer 6

看看 Django Poor Man 的 Cron，它是一个 Django 应用程序，它利用垃圾邮件机器人、搜索引擎索引机器人等来大约定期运行计划任务

请参阅：http://code.google.com/p/django-poormanscron/

Answer 7

不久前我有完全相同的要求，最终使用 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

希望这对某人有帮助！

Answer 8

Brian Neal 通过 cron 运行管理命令的建议效果很好，但如果你正在寻找更强大的东西（但不像 Celery 那样复杂），我会考虑像 克罗诺斯：

# app/cron.py

import kronos

@kronos.register('0 * * * *')
def task():
    pass

Answer 9

用于调度程序作业的 Django APScheduler。 Advanced Python Scheduler (APScheduler) 是一个 Python 库，可让您安排 Python 代码稍后执行，可以仅执行一次或定期执行。 您可以根据需要随时添加新职位或删除旧职位。

注意：我是这个库的作者

安装APScheduler

pip install apscheduler

查看文件函数调用

文件名：scheduler_jobs.py

def FirstCronTest():
    print("")
    print("I am executed..!")

配置调度程序

makeexecute.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

Answer 10

RabbitMQ 和 Celery 比 Cron 有更多的功能和任务处理能力。 如果任务失败不是问题，并且您认为将在下一次调用中处理损坏的任务，那么 Cron 就足够了。

芹菜和 AMQP 将让您处理损坏的任务，并且它将由另一个工作人员（Celery工作人员）再次执行监听下一个要处理的任务），直到达到任务的 max_retries 属性。 您甚至可以在失败时调用任务，例如记录失败，或在达到 max_retries 后向管理员发送电子邮件。

当您需要扩展应用程序时，您可以分发 Celery 和 AMQP 服务器。

Answer 11

我个人使用 cron，但是 作业调度 部分href="https://github.com/django-extensions/django-extensions" rel="noreferrer">django-extensions 看起来很有趣。

Answer 12

尽管不是 Django 的一部分，Airflow 是一个更新的项目（截至 2016 年）对于任务管理很有用。

Airflow 是一个工作流自动化和调度系统，可用于创作和管理数据管道。 基于 Web 的 UI 为开发人员提供了一系列用于管理和查看这些管道的选项。

Airflow 用 Python 编写，并使用 Flask 构建。

Airflow 由 Airbnb 的 Maxime Beauchemin 创建，并于 2015 年春季开源。它于 2016 年冬季加入 Apache 软件基金会的孵化计划。这是 Git 项目页面 和一些附加背景信息。

Answer 13

将以下内容放在 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

Answer 14

我只是想到了这个相当简单的解决方案：

1. 定义一个视图函数 do_work(req, param) 就像使用任何其他视图一样，使用 URL 映射，返回 HttpResponse 等等。
2. 使用您的计时首选项（或使用 AT 或 Windows 中的计划任务）设置运行 curl http://localhost/your/mapped/url?param=value。

您可以添加参数，但只需将参数添加到 URL 中即可。

告诉我你们在想什么。

[更新] 我现在正在使用 django-extensions 而不是卷曲。

我的 cron 看起来像这样：

@hourly python /path/to/project/manage.py runjobs hourly

...每天、每月等等。 您还可以将其设置为运行特定作业。

我发现它更易于管理且更清洁。 不需要将 URL 映射到视图。 只需定义您的工作类别和 crontab 即可。

Answer 15

在代码部分之后，我可以写任何东西，就像我的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/

Answer 16

你绝对应该看看 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,
},
}

Answer 17

是的，上面的方法太棒了。 我尝试了其中一些。 最后，我找到了这样的方法：

    from threading import Timer

    def sync():

        do something...

        sync_timer = Timer(self.interval, sync, ())
        sync_timer.start()

就像递归一样。

好的，希望这个方法能够满足您的要求。 :)

Answer 18

一个更现代的解决方案（与 Celery 相比）是 Django Q：
https://django-q.readthedocs.io/en/latest/index.html html

它有很棒的文档并且很容易理解。 缺乏 Windows 支持，因为 Windows 不支持进程分叉。 但如果您使用 Windows for Linux 子系统创建开发环境，它就可以正常工作。

Answer 19

我今天也遇到了和你类似的问题。

我不想让服务器通过 cron 来处理它（并且大多数库最终只是 cron 助手）。

所以我创建了一个调度模块并将其附加到 init 。

这不是最好的方法，但它帮助我将所有代码放在一个位置，并且其执行与主应用程序相关。

Answer 20

我使用 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

Answer 21

我不确定这对任何人都有用，因为我必须提供系统的其他用户来安排作业，而不让他们访问实际的服务器（Windows）任务计划程序，我创建了这个可重用的应用程序。

请注意，用户有权访问服务器上的一个共享文件夹，他们可以在其中创建所需的命令/任务/.bat 文件。 然后可以使用此应用程序安排此任务。

应用程序名称是 Django_Windows_Scheduler

屏幕截图：

Answer 22

如果您想要比 Celery 更可靠的东西，请尝试构建在AWS SQS/SNS之上的TaskHawk。

请参阅：http://taskhawk.readthedocs.io

Answer 23

对于简单的码头化项目，我无法真正看到任何现有的答案适合。

因此，我编写了一个非常简单的解决方案，无需外部库或触发器，它可以自行运行。 不需要外部 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',
    ...
]

Answer 24

简单的方法是编写自定义 shell 命令，请参阅 Django 文档 并在 Linux 上使用 cronjob 执行它。 不过，我强烈建议使用像 RabbitMQ 这样的消息代理和 celery。 也许你可以看看
这个教程

Answer 25

一种替代方法是使用 Rocketry：

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 本质上也是如此只是一个可以生成异步任务、线程和进程的复杂循环。

免责声明：我是作者。

Answer 26

另一种选择，类似于 Brian Neal 的回答，使用 RunScripts

然后你不需要设置命令。 这样做的优点是文件夹结构更灵活或更清晰。

该文件必须实现 run() 函数。 这是运行脚本时所调用的。 您可以导入 django 项目的任何模型或其他部分以在这些脚本中使用。

然后，就

python manage.py runscript path.to.script