Flask-APScheduler詳細配置使用

1.配置

from App.tasks.DatabaseTask import send_ding_test  # 我的任務函數
from flask_apscheduler.auth import HTTPBasicAuth
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore


class Config(object):

    JOBS = [
        # interval定時執行（從start_date到end_date，間隔20s，包含首尾）
        # func也可以寫字符串形式，例如：'App.tasks.DatabaseTask:send_ding_test'
        {
            'id': 'job2',
            'func': send_ding_test,
            'trigger': 'interval',
            'start_date': '2021-01-27 13:31:00',
            'end_date': '2021-01-27 13:33:00',
            'seconds': 20,
            'replace_existing': True  # 重新執行程序時，會將jobStore中的任務替換掉
        },
        # date一次執行
        {
            'id': 'job1',
            'func': send_ding_test,
            'trigger': 'date',
            'run_date': '2021-01-30 11:22:00',
            'replace_existing': True
        },
        # cron式定時調度，類似linux的crontab
        {
            'id': 'job3',
            'func': send_ding_test,
            'trigger': 'cron',
            'day_of_week': '0-6',
            'month': '*',
            'hour': '6',
            'minute': '0',
            'second': '0',
            'replace_existing': True
        }

    ]

    # 存儲定時任務（默認是存儲在內存中）  
    SCHEDULER_JOBSTORES  = {'default':SQLAlchemyJobStore(url='mysql+pymysql://xxx/xx')}
    # 設置時區，時區不一致會導致定時任務的時間錯誤
    SCHEDULER_TIMEZONE = 'Asia/Shanghai'
    # 一定要開啟API功能，這樣才可以用api的方式去查看和修改定時任務
    SCHEDULER_API_ENABLED = True
    # api前綴（默認是/scheduler）
    SCHEDULER_API_PREFIX = '/scheduler'
    # 配置允許執行定時任務的主機名
    SCHEDULER_ALLOWED_HOSTS = ['*']
    # auth驗證。默認是關閉的，
    SCHEDULER_AUTH = HTTPBasicAuth()
    # 設置定時任務的執行器（默認是最大執行數量為10的線程池）
    SCHEDULER_EXECUTORS = {'default': {'type': 'threadpool', 'max_workers': 10}}
    # 另外flask-apscheduler內有日志記錄器。name為apscheduler.scheduler和apscheduler.executors.default。如果需要保存日志，則需要對此日志記錄器進行配置
    

 

2.初始化app

import logging
from flask import Flask
from config import Config  # 上邊的配置文件
from flask_apscheduler.scheduler import APScheduler
from logging.handlers import RotatingFileHandler


# flask-apscheduler內置有日志器，為了讓內部的日志器打印的內容輸出，我這里做了個配置
# 創建日志記錄器，指明日志保存路徑，每個日志大小，保存日志文件個數上限
file_log_handler = RotatingFileHandler('logs/runserver.log', maxBytes=1024*1024*100, backupCount=5)
# 創建日志的記錄格式,]，日志等級，記錄時間，報錯位置，行數，日志信息
formatter = logging.Formatter(
    fmt="%(asctime)s - %(levelname)s - %(threadName)s:%(thread)s - %(filename)s - %(funcName)s - %(message)s",
    datefmt='%Y-%m-%d %H:%M:%S %a'
)
# 為剛創建的日志記錄器設置日志記錄格式
file_log_handler.setFormatter(formatter)
# 為全局日志對象添加日志記錄器
logger = logging.getLogger("apscheduler")
logger.addHandler(file_log_handler)
logger.setLevel(logging.INFO)



def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.start()
    
    # 配置api權限驗證的回調函數
    @scheduler.authenticate
    def authenticate(auth):
        return auth['username'] == 'guest' and auth['password'] == 'guest'
    
    
app = create_app()
app.run(host='127.0.0.1', port='7788')

3.API調用

Flask-APScheduler內置了豐富的api接口，可以讓開發者動態的查看和更改定時任務，非常方便。這里我找到內部的一部分源碼，可以看到所有的api接口的調用方式。

http://xxxxx/scheduler

    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')

　　帶權限驗證

import base64

# 用戶名密碼用:分隔
string = "guest:guest".encode('utf-8')

a = base64.b64encode(string)
# print(a)
# b'Z3Vlc3Q6Z3Vlc3Q='
print(b"basic " + a)  # 最后必須拼接上basic
# b'basic Z3Vlc3Q6Z3Vlc3Q='