airflow 定时任务 python flask 定时任务方案

目录

一、安装Flask-APScheduler

二、基本概念

三、使用步骤

四、使用实例

五、RuntimeError: Working outside of application context

六、flask_apscheduler Unable to determine the name of the local timezone

---

APScheduler是基于Quartz的一个Python定时任务框架。提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务。
在线文档：https://apscheduler.readthedocs.io/en/latest/userguide.html

一、安装Flask-APScheduler

pip install Flask-APScheduler

二、基本概念

APScheduler有四大组件：
1、触发器 triggers ：
触发器包含调度逻辑。每个作业都有自己的触发器，用于确定下一个任务何时运行。除了初始配置之外，触发器是完全无状态的。
有三种内建的trigger:
（1）date: 特定的时间点触发
（2）interval: 固定时间间隔触发
（3）cron: 在特定时间周期性地触发

interval 触发器 
参数 说明 

weeks (int) 间隔几周 
days (int) 间隔几天 
hours (int) 间隔几小时 
minutes (int) 间隔几分钟 
seconds (int) 间隔多少秒 
start_date (datetime 或 str) 开始日期 
end_date (datetime 或 str) 结束日期 
timezone (datetime.tzinfo 或str) 时区

每隔两分钟执行一次 job_func 方法

scheduler .add_job(job_func, 'interval', minutes=2)

在 2017-12-13 14:00:01 ~ 2017-12-13 14:00:10 之间, 每隔两分钟执行一次 job_func 方法

scheduler .add_job(job_func, 'interval', minutes=2, start_date='2017-12-13 14:00:01' , end_date='2017-12-13 14:00:10')

cron 触发器: 
参数 说明 
year (int 或 str) 年，4位数字 
month (int 或 str) 月 (范围1-12) 
day (int 或 str) 日 (范围1-31 
week (int 或 str) 周 (范围1-53) 
day_of_week (int 或 str) 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun) 
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) 指定时区

在每年 1-3、7-9 月份中的每个星期一、二中的 00:00, 01:00, 02:00 和 03:00 执行 job_func 任务

scheduler .add_job(job_func, 'cron', month='1-3,7-9',day='0, tue', hour='0-3')

2、任务储存器 job stores：用于存放任务，把任务存放在内存（为默认MemoryJobStore）或数据库中。
3、执行器 executors： 执行器是将任务提交到线程池或进程池中运行，当任务完成时，执行器通知调度器触发相应的事件。
4、调度器 schedulers： 把上方三个组件作为参数，通过创建调度器实例来运行

根据开发需求选择相应的组件，下面是不同的调度器组件：
BlockingScheduler 阻塞式调度器：适用于只跑调度器的程序。
BackgroundScheduler 后台调度器：适用于非阻塞的情况，调度器会在后台独立运行。
AsyncIOScheduler AsyncIO调度器，适用于应用使用AsnycIO的情况。
GeventScheduler Gevent调度器，适用于应用通过Gevent的情况。
TornadoScheduler Tornado调度器，适用于构建Tornado应用。
TwistedScheduler Twisted调度器，适用于构建Twisted应用。
QtScheduler Qt调度器，适用于构建Qt应用。

三、使用步骤

1、新建一个调度器schedulers
2、添加调度任务
3、运行调度任务

四、使用实例

# -*- coding: utf-8 -*-

# @Time : 2020/12/23 下午1:57
# @File : flask_cheduler
# @Software: PyCharm


from flask import Flask, jsonify
from flask_apscheduler import APScheduler
from datetime import datetime
app = Flask(__name__)
scheduler = APScheduler()


def my_job():
    print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))



@app.route('/update')
def update():
    my_job()
    return jsonify({'code': 0, 'data': 'update'})

@app.route('/extraction', methods=['post'])
def extraction():
    return jsonify({'code': 0, 'data': 'extraction'})



if __name__ == "__main__":
    # 定时任务 
    scheduler.add_job(func=my_job,id='1',trigger='cron', month='*/1')
    scheduler.init_app(app=app)
    scheduler.start()
    app.run(debug=False, host='0.0.0.0',port=10002)

五、RuntimeError: Working outside of application context

问题的原因：在没有激活程序上下文之前进行了一些程序上下文或请求上下文的操作

解决办法：推送程序上下文，在获得程序上下文后再执行相应的操作

from flask import Flask, jsonify, current_app
app = Flask(__name__)
app_ctx = app.app_context()  
# app.app_context()是一个上下文表达式，它返回了一个上下文管理器AppContext() 

app_ctx.push()

print(current_app.name)  # 要执行的代码

# app_ctx.pop()

或 

from flask import Flask, jsonify, current_app
app = Flask(__name__)
with app.app_context():
  #TODO  要执行的代码

六、flask_apscheduler Unable to determine the name of the local timezone

在Linux下配置flask_apscheduler遇到了问题：

ValueError: Unable to determine the name of the local timezone – you
must explicitly specify the name of the local timezone. Please refrain
from using timezones like EST to prevent problems with daylight saving
time. Instead, use a locale based timezone name (such as
Europe/Helsinki).

解决方法是添加环境变量：

import os
os.environ['TZ']= 'Asia/Shanghai'