快捷導(dǎo)航

Python flask框架定時(shí)任務(wù)apscheduler應(yīng)用介紹

更新時(shí)間：2022年10月10日 15:14:16 作者：HHYZBC

Flask是Python社區(qū)非常流行的一個(gè)Web開(kāi)發(fā)框架，本文將嘗試將介紹APScheduler應(yīng)用于Flask之中實(shí)現(xiàn)定時(shí)任務(wù)，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧

基本使用

flask-apscheduler的相關(guān)配置，我們會(huì)將它和其它擴(kuò)展一起，放在應(yīng)用的配置里

class Config(object):
    // 配置項(xiàng)
    JOBS = [
        {
            'id': 'job1',
            'func': 'run:add',
            'args': (1, 2),
            'trigger': 'interval',
            'seconds': 3
        }
    ]
    SCHEDULER_API_ENABLED = True
def add(a, b):
    print(a+b)

JOBS列表的每一個(gè)元素表示一個(gè)定時(shí)任務(wù)，列子中只有一個(gè)interval任務(wù)，表示每隔3秒運(yùn)行一次函數(shù)add。func指定調(diào)用的函數(shù)，args表示傳入函數(shù)的參數(shù)，trigger表示啟動(dòng)方式，常用的有兩種，分別是trigger和cron。

上邊我們?cè)O(shè)置了SCHEDULER_API_ENABLED = True，可以通過(guò)訪問(wèn)http://127.0.0.1:5000/scheduler，其中scheduler是默認(rèn)的RESTful API前綴

通過(guò)查看源碼，可以發(fā)現(xiàn)flask-apscheduler提供了以下的接口

def _load_api(self):
    """
    Add the routes for the scheduler API.
    """
    self._add_url_route('get_scheduler_info', '', api.get_scheduler_info, 'GET')
    self._add_url_route('add_job', '/jobs', api.add_job, 'POST')
    self._add_url_route('get_job', '/jobs/<job_id>', api.get_job, 'GET')
    self._add_url_route('get_jobs', '/jobs', api.get_jobs, 'GET')
    self._add_url_route('delete_job', '/jobs/<job_id>', api.delete_job, 'DELETE')
    self._add_url_route('update_job', '/jobs/<job_id>', api.update_job, 'PATCH')
    self._add_url_route('pause_job', '/jobs/<job_id>/pause', api.pause_job, 'POST')
    self._add_url_route('resume_job', '/jobs/<job_id>/resume', api.resume_job, 'POST')
    self._add_url_route('run_job', '/jobs/<job_id>/run', api.run_job, 'POST')

如果需要查看當(dāng)前運(yùn)行的所有定時(shí)任務(wù)，則請(qǐng)求http://127.0.0.1:5000/scheduler/jobs即可。

trigger啟動(dòng)方式

trigger表示間隔啟動(dòng)，在trigger方式中，使用seconds配置間隔多久啟動(dòng)一次，單位是秒。

cron啟動(dòng)方式

cron表示定時(shí)啟動(dòng)

class Config(object):
    JOBS = [
        {
            'id': 'job1',
            'func': 'scheduler:task',
            'args': (1, 2),
            'trigger': 'cron',
            'day': '*',
            'hour': '13',
            'minute': '16',
            'second': '20'
        }
    ]
    SCHEDULER_API_ENABLED = True
def task(a, b):
    print(str(datetime.datetime.now()) + ' execute task ' + '{}+{}={}'.format(a, b, a + b))

該配置項(xiàng)則表示每天的13點(diǎn)16分20秒啟動(dòng)一次。*表示全部。

有關(guān)常用的cron配置有：

day

表示天

hour

表示小時(shí)

minute

表示分鐘

second

表示秒

week

周

day_of_week

星期幾，如星期天使用sun，星期五使用fri，其他的類似。

使用裝飾器定時(shí)啟動(dòng)任務(wù)

from flask import Flask
from flask_apscheduler import APScheduler
import datetime
class Config(object):
    SCHEDULER_API_ENABLED = True
scheduler = APScheduler()
# interval examples
@scheduler.task('interval', id='do_job_1', seconds=30, misfire_grace_time=900)
def job1():
    print(str(datetime.datetime.now()) + ' Job 1 executed')

表示每隔30秒調(diào)用一次job1函數(shù)。

到此這篇關(guān)于Python flask框架定時(shí)任務(wù)apscheduler應(yīng)用介紹的文章就介紹到這了,更多相關(guān)Python flask apscheduler內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: