Flask-APscheduler使用チュートリアル

10631 ワード
Python
Flask-APscheduler使用チュートリアル
作者:lizhonglingithub:https://github.com/Leezhonglin/blog: https://leezhonglin.github.io/
APSchedulerはPythonライブラリで、後で実行するPythonコードを手配することができます.タスクスケジューリングフレームワークで、タイミングタスクコントローラとして使用でき、削除タスクを追加することができます.ジョブがデータベースに格納されている場合は、スケジューラが再起動した後も実行され、ステータスが維持されます.スケジューラを再起動すると、オフラインで実行すべきすべてのジョブが実行されます.
また、APSchedulerは、アプリケーション固有の代替プログラムとして、cronデーモンやWindowsタスクスケジューラなどのプラットフォーム間で使用することもできます.ただし、APScheduler自体はデーモンやサービスではなく、コマンドラインツールも付属していないことに注意してください.主に既存のアプリケーションで実行するために使用されます.つまり、APSchedulerは、スケジューラ・サービスを構築したり、専用のスケジューラ・プロセスを実行したりするためのいくつかの構築ブロックを提供しています.
APSchedulerには、3つの組み込みスケジューリングシステムがあります.
  • Cronスケジューリング(オプションの開始/終了時間)
  • 区間に基づく実行(偶数間隔運転ジョブ、オプションの開始/終了時間)
    • .
  • 一括遅延実行(設定日時に1回のジョブを実行)
    • .APSchedulerは、pythonのバックグラウンドプログラムを提供するためのpythonのサードパーティライブラリです.次の4つのコンポーネントがあります.
  • triggers:タスクトリガコンポーネント、タスクトリガ方式
    • を提供する
  • job stores:タスクストアコンポーネント、タスク保存方式
    • を提供
  • executors:タスクスケジューリングコンポーネント、タスクスケジューリング方式
    • を提供する
  • schedulers:タスクスケジューリングコンポーネント、タスク動作方式
    • を提供する
    具体的な内容は以下の通りです.
    triggers:3つのタスクトリガ方式をサポート
  • date:固定日付トリガ、タスクは1回しか実行されず、実行が完了すると自動的にクリアされます.指定した実行時間を逃すと、タスクは作成されません.
    パラメータ
    説明
    run_date(datetimeまたはstr)
    ジョブの実行日時
    timezone(datetime.tzinfoまたはstr)
    タイムゾーンの指定
      #   2019-4-24 00:00:01        start_system   
scheduler .add_job(start_system, 'date', run_date='2019-4-24 00:00:01', args=['text'])
  • interval:時間間隔トリガで、一定の時間間隔ごとに1回実行されます.
    パラメータ
    説明
    weeks (int)
    数週間間隔
    days (int)
    数日おきに
    hours (int)
    数時間間隔
    minutes (int)
    数分間隔
    seconds (int)
    何秒間隔
    start_date(datetimeまたはstr)
    開始日
    end_date(datetimeまたはstr)
    終了日
    #   2019-4-24 00:00:00 - 2019-4-24 08:00:00   ,           alarm_job   
scheduler .add_job(alarm_job, 'interval', hours=2, start_date='2019-4-24 00:00:00' , end_date='2019-4-24 08:00:00')
  • cron:cronスタイルのタスクトリガ
    パラメータ
    説明
    year(intまたはstr)
    4桁を示す年(2019)
    month(int|str)
    月(範囲1-12)
    day(int|str)
    日(範囲1-31)
    week(int|str)
    週(範囲1～53)
    day_of_week (int|str)
    1週間の何日目を表すかは、0-6でも英語でも略して表すことができます.
    hour (int|str)
    表示取値範囲が0-23の場合
    minute (int|str)
    表示値範囲0-59分
    second (int|str)
    値範囲が0～59秒であることを示します
    start_date (datetime|str)
    開始時間を表す
    end_date (datetime|str)
    終了時間を表す
    timezone (datetime.tzinfo|str)
    タイムゾーン値の表示
    (int|str)表示パラメータは、intタイプであってもよいし、strタイプ(datetime|str)表示パラメータは、datetimeタイプであってもよいし、strタイプであってもよい
    例えば、5秒ごとにプログラムが実行されることを示し、interval間隔スケジューリングにおけるseconds=5 
    sched.add_job(my_job, 'cron',second = '*/5')
    • に相当する.
    job stores:4つのタスクストレージ方式をサポート
  • memory:デフォルトの構成タスクはメモリに存在する
  • mongdb:ドキュメントデータベースストレージ
    • をサポート
  • sqlalchemy:リレーショナル・データベース・ストレージ
    • をサポート
  • redis:データベースに対するキー値の格納をサポートする
    • schedulers:スケジューラは主に3種類に分けられ、1つは独立して実行され、1つはバックグラウンドで実行され、最後の1つは他のプログラムと協力して使用されます.
  • BlockingScheduler:このスケジューラがあなたのアプリケーションの のものである場合、
    • を使用します.
  • BackgroundScheduler: のときに使用し、 の
    • でタスクを実行します.
  • AsyncIOScheduler:プログラムが IO の場合
    • を使用します.
  • GeventScheduler:geventフレームと組み合わせて
    • を使用
  • TornadoScheduler:tornadoフレームと組み合わせて
    • を使用
  • TwistedScheduler:Twistedフレームと組み合わせて
    • を使用
  • QtScheduler:qtアプリケーション開発時に
    • を使用
    1.インストール
    pip install Flask-APScheduler

    2.flaskでのスケジューラの使用
    まずはflaskのcreate_app関数で最初のスケジューラを作成する
    from apscheduler.triggers.interval import IntervalTrigger

###        app
def create_app(config_name):
    app=Flask(__name__)
    CORS(app)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    ###      
    db.init_app(app)
#####################################################      
  #        
  #            
    scheduler.init_app(app)
    # #   FLASK DEBUG          
    if os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
        scheduler.api_enabled = True
        scheduler.init_app(app)
		#    interval  ，                interval  
        interval = IntervalTrigger(
        	days = 2,
        	start_date='2019-4-24 08:00:00',
        	end_date='2099-4-24 08:00:00',
        	timezone='Asia/Shanghai')
        # dbDump            ，             
        scheduler.add_job(func=dbDump,trigger=interval,id='bak_one')
        scheduler.start()
######################################################      

    ###            
    # init_static_path()
    ###    
    register_app(app)
    # ###     
    init_log()
    return app

    上記のタスクを再置換する必要がある場合は、タスクにパラメータreplace_existing=Trueを追加して、上記の2日間のバックアップから毎日1回のバックアップに変更できます.構成に成功すると、上のタイマーを自動的に置き換えます.
     #                
if os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
	interval = IntervalTrigger(
    days=1,
    start_date='2019-4-24 08:00:00',
    end_date='2099-4-24 08:00:00',
    timezone='Asia/Shanghai')
	scheduler.add_job(
    func=dbDump,
		trigger=interval,
		id='bak_one', 
		replace_existing=True)

    replace_existing=Trueというパラメータは非常に重要で、この場所で長い間やっていたので、最初のタスクを置き換えることができませんでした.
    問題解決:
    scheduler.add_job(dbDump,'interval',id='bak_one',days=1,start_date='2019-4-24 08:00:00',end_date='2099-4-24 08:00:00',replace_existing=True)

    上記のように作成したタスクであればpyinstallerを使用して生成した実行可能ファイルをパッケージ化し、実行時に次のような問題が発生します.
    APScheduler: LookupError: No trigger by the name “interval” was found

    解決策は、私の代わりにタスクを作成すれば解決できます.
    C++ポインタ(Pointer)の理解(ポインタを指すポインタ、参照を指すポインタ)
    Ansibleの呼吸 参ノ型 フィルター芸「リスト内にある重複している辞書の数を確認」