安排在Flask上每小時運行一個函數

Question

我有一個Flask虛擬主機，無法訪問cron命令。 如何每小時執行一次Python函數？

Answer 1

要更新肖恩·維埃拉的answer：由於Scheduler()在APScheduler V3.0刪除，我們現在可以（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") 

注意，其中兩個調度程序將被啓動時，燒瓶在調試模式下。有關更多信息，請查看this問題。

Answer 2

你可以利用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() 

Answer 3

您可能希望使用與調度一些隊列機制像RQ scheduler或一些更重的像芹菜（最有可能一個矯枉過正）。

Answer 4

另一種方法是使用燒瓶APScheduler與瓶很好的發揮，如：

• 負載從瓶配置調度配置，
• 負荷作業定義從瓶配置

更多information here：

https://pypi.python.org/pypi/Flask-APScheduler

Answer 5

您可以嘗試使用APScheduler's BackgroundScheduler將間隔作業集成到Flask應用程序中。下面是使用藍圖和應用工廠（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() 

希望它能幫助:)

編號：

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

Answer 6

我在應用程序調度程序的概念上有點新，但是我在這裏發現的APScheduler v3.3.1，有點不同。我認爲，最新的版本中，封裝結構，類名稱等，都變了，所以我把這裏我最近做了一個新的解決方案，集成了基本的應用瓶：

#!/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() 

我如果有人對這個例子的更新感興趣，也可以離開這個Gist here。

這裏有一定的參考，爲未來的讀數：

• APScheduler文件：https://apscheduler.readthedocs.io/en/latest/
• 守護= TRUE：https://docs.python.org/3.4/library/threading.html#thread-objects