安排一个功能在Flask上每小时运行一次

I have a Flask web hosting with no access to cron command. How can I execute some Python function every hour?

我有一个Flask web托管，无法访问cron命令。我怎样才能每小时执行一些Python函数？

6 个解决方案

#1

To update Sean Vieira's answer: Since Scheduler() was removed in APScheduler v3.0, we can now (v3.2) use BackgroundScheduler():

更新Sean Vieira的答案：由于在APScheduler v3.0中删除了Scheduler（），我们现在可以（v3.2）使用BackgroundScheduler（）：

import time
import atexit

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger

scheduler = BackgroundScheduler()
scheduler.start()
scheduler.add_job(
    func=print_date_time,
    trigger=IntervalTrigger(seconds=5),
    id='printing_job',
    name='Print date and time every five seconds',
    replace_existing=True)
# Shut down the scheduler when exiting the app
atexit.register(lambda: scheduler.shutdown())

def print_date_time():
    print time.strftime("%A, %d. %B %Y %I:%M:%S %p")

Note that two of these schedulers will be launched when Flask is in debug mode. For more information, check out this question.

请注意，当Flask处于调试模式时，将启动其中两个调度程序。有关更多信息，请查看此问题。

#2

You could make use of APScheduler in your Flask application and run your jobs via its interface:

您可以在Flask应用程序中使用APScheduler并通过其界面运行您的作业：

import atexit

from apscheduler.scheduler import Scheduler
from flask import Flask

app = Flask(__name__)

cron = Scheduler(daemon=True)
# Explicitly kick off the background thread
cron.start()

@cron.interval_schedule(hours=1)
def job_function():
    # Do your work here


# Shutdown your cron thread if the web process is stopped
atexit.register(lambda: cron.shutdown(wait=False))

if __name__ == '__main__':
    app.run()

#3

I'm a little bit new with the concept of application schedulers, but what I found here for APScheduler v3.3.1 , it's something a little bit different. I believe that for the newest versions, the package structure, class names, etc., have changed, so I'm putting here a fresh solution which I made recently, integrated with a basic Flask application:

我对应用程序调度程序的概念有点新意，但我在APScheduler v3.3.1中找到了它，它有点不同。我相信对于最新版本，包结构，类名等已经改变，所以我在这里提出了一个我最近制作的新解决方案，与基本的Flask应用程序集成：

#!/usr/bin/python3
""" Demonstrating Flask, using APScheduler. """

from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask

def sensor():
    """ Function for test purposes. """
    print("Scheduler is alive!")

sched = BackgroundScheduler(daemon=True)
sched.add_job(sensor,'interval',minutes=60)
sched.start()

app = Flask(__name__)

@app.route("/home")
def home():
    """ Function for test purposes. """
    return "Welcome Home :) !"

if __name__ == "__main__":
    app.run()

I'm also leaving this Gist here, if anyone have interest on updates for this example.

如果有人对此示例的更新感兴趣，我也会在这里留下这个要点。

Here are some references, for future readings:

以下是一些参考资料，供将来阅读：

APScheduler Doc: https://apscheduler.readthedocs.io/en/latest/
APScheduler Doc：https：//apscheduler.readthedocs.io/en/latest/
daemon=True: https://docs.python.org/3.4/library/threading.html#thread-objects
daemon = True：https：//docs.python.org/3.4/library/threading.html#thread-objects

#4

Another alternative might be to use Flask-APScheduler which plays nicely with Flask, e.g.:

另一种选择可能是使用Flask-APScheduler，它可以很好地与Flask一起使用，例如：

Loads scheduler configuration from Flask configuration,
从Flask配置加载调度程序配置，
Loads job definitions from Flask configuration
从Flask配置加载作业定义

More information here:

#5

You could try using APScheduler's BackgroundScheduler to integrate interval job into your Flask app. Below is the example that uses blueprint and app factory (init.py) :

您可以尝试使用APScheduler的BackgroundScheduler将间隔作业集成到Flask应用程序中。以下是使用blueprint和app factory（init.py）的示例：

from datetime import datetime

# import BackgroundScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask

from webapp.models.main import db 
from webapp.controllers.main import main_blueprint    

# define the job
def hello_job():
    print('Hello Job! The time is: %s' % datetime.now())

def create_app(object_name):
    app = Flask(__name__)
    app.config.from_object(object_name)
    db.init_app(app)
    app.register_blueprint(main_blueprint)
    # init BackgroundScheduler job
    scheduler = BackgroundScheduler()
    # in your case you could change seconds to hours
    scheduler.add_job(hello_job, trigger='interval', seconds=3)
    scheduler.start()

    try:
        # To keep the main thread alive
        return app
    except:
        # shutdown if app occurs except 
        scheduler.shutdown()

Hope it helps :)

希望能帮助到你：）

Ref :

参考：

https://github.com/agronholm/apscheduler/blob/master/examples/schedulers/background.py
https://github.com/agronholm/apscheduler/blob/master/examples/schedulers/background.py

#6

You might want to use some queue mechanism with scheduler like RQ scheduler or something more heavy like Celery (most probably an overkill).

您可能希望使用一些队列机制与调度程序（如RQ调度程序）或更重的东西（如Celery）（最有可能是一种过度杀伤力）。

#1