Flask-定时任务

这篇具有很好参考价值的文章主要介绍了Flask-定时任务。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

一、简单使用

二、apscheduler 

triggers

job stores

schedulers

使用上下文

日志设置


pip install Flask-APScheduler

 Flask-APScheduler: Tips - Flask APScheduler

Flask-APScheduler 是基于 python 第三方库 apscheduler 做的集成, 所以官网上只有一些简单的使用案例,详细的配置还是要看 apscheduler的文档。

apscheduler: 

Advanced Python Scheduler — APScheduler 3.9.1 documentation

  引用:

Flask-APScheduler使用教程 | Mr. Lee's blog

一、简单使用

__init__.py:

from flask import Flask

app = Flask(__name__)
app.config.from_object('config.config')

# 初始化db
db = SQLAlchemy()
db.init_app(app)


# 初始化定时任务
from models.taskSchedule import scheduler
scheduler.init_app(app)
scheduler.start()
# 修改调度器执行组件冗余日志级别
logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING)

taskSchedule.py:

from flask_apscheduler import APScheduler


scheduler = APScheduler()



# interval example, 间隔执行, 每30秒执行一次
@scheduler.task('interval', id='do_job_1', seconds=30, misfire_grace_time=900)
def job1():
    print('Job 1 executed')


# cron examples, 每分钟执行一次
@scheduler.task('cron', id='do_job_2', minute='*')
def job2():
    print('Job 2 executed')

# 每周执行一次
@scheduler.task('cron', id='do_job_3', week='*', day_of_week='sun')
def job3():
    print('Job 3 executed')

也可以在程勋运行后添加定时任务:

scheduler.start()
scheduler.add_job(**args)

二、apscheduler 

apscheduler 四个组件:

  • triggers: 任务触发器组件,提供任务触发方式
  • job stores: 任务商店组件,提供任务保存方式
  • executors: 任务调度组件,提供任务调度方式
  • schedulers: 任务调度组件,提供任务工作方式

triggers

支持三种任务触发方式

  • date:

固定日期触发器,任务只运行一次,运行完毕自动清除;若错过指定运行时间,任务不会被创建

| 参数 | 说明 |
| :——————————– | :——————- |
| 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:

时间间隔触发器,每个一定时间间隔执行一次。

| 参数 | 说明 |
| —————————- | ———- |
| 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) 表示四位数的年份 (2019)
month(int\ str) 月 (范围1-12)
day(int\ str) 日 (范围1-31)
week(int\ str) 周 (范围1-53)
day_of_week (int\ str) 表示一周中的第几天,既可以用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

支持四种任务存储方式

  • memory:默认配置任务存在内存中
  • mongdb:支持文档数据库存储
  • sqlalchemy:支持关系数据库存储
  • redis:支持键值对数据库存储

schedulers

调度器主要分三种,一种独立运行的,一种是后台运行的,最后一种是配合其它程序使用

  • BlockingScheduler: 当这个调度器是你应用中 唯一要运行 的东西时使用
  • BackgroundScheduler: 当 不运行其它框架 的时候使用,并使你的任务在 后台运行
  • AsyncIOScheduler: 当你的程序是 异步IO模型 的时候使用
  • GeventScheduler: 和 gevent 框架配套使用
  • TornadoScheduler: 和 tornado 框架配套使用
  • TwistedScheduler: 和 Twisted 框架配套使用
  • QtScheduler: 开发 qt 应用的时候使用

Flask-APScheduler 中默认使用的就是 BackgroundScheduler:

scheduler.py:

#.....

class APScheduler(object):
    """Provides a scheduler integrated to Flask."""

    def __init__(self, scheduler=None, app=None):
        self._scheduler = scheduler or BackgroundScheduler()
        self._host_name = socket.gethostname().lower()
        self._authentication_callback = None


# .....

使用上下文

如果正在使用 Flask-SQLAlchemy 并在定时任务中执行数据库操作,需要提供 Flask 应用程序上下文:

from flask_apscheduler import APScheduler
scheduler = APScheduler()
 
@scheduler.task(
     "interval",
      id="update_news_graph_job",
     minutes = 16
 )
def update_news_graph():

        # 提供flask上下文对象
        with scheduler.app.app_context():
            result = db.session.execute(query_sql)
当然, scheduler 需要在flask中注册:
scheduler.init_app(app)
scheduler.start()

官网关于上下文的解释:

Basic Application - Flask APScheduler

Introduction into Contexts — Flask-SQLAlchemy Documentation (2.x)

日志设置

如果定时任务执行间隔几秒钟, 调度程序的日志会很多,可以设置调度程序日志级别或完全禁用:文章来源地址https://www.toymoban.com/news/detail-403209.html

#设置调度程序的日志级别, 原本级别为info

scheduler.start()
scheduler.add_job(every_minute, trigger='cron', second=0, id='every_minute')
logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING)


#或者禁用调度程序日志
logging.getLogger('apscheduler.executors.default').propagate = False

到了这里,关于Flask-定时任务的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Django - 定时任务框架【django-apscheduler】基本使用详解(二)

    一个网页会有很多数据是不需要经常变动的,比如说首页,变动频率低而访问量大,我们可以把它静态化,这样就不需要每次有请求都要查询数据库再返回,可以减少服务器压力 我们可以使用Django的模板渲染功能完成页面渲染 APScheduler的全称是Advanced Python Scheduler。它是一个

    2024年02月12日
    浏览(45)
  • Django之定时任务--apscheduler

    在写一个兴趣班教务管理的小程序时,使用了Django后台来管理数据库,由于每天要根据现有设定的班级进行课表的更新,所以需要一个定时器,但是开发环境是在windows上,后台部署在linux上,问了chatGPT,说是apscheduer支持跨平台使用,然后就选了这个,以下是使用过程,按c

    2024年02月13日
    浏览(40)
  • flask python 设置定时任务 flask 周期性执行任务方案

    flask 通常使用 flask_apscheduler 框架设计定时任务,flask_apscheduler 功能很全面,能按设定的时间规则执行任务,可以持久化到各类数据库(mysql,redis,mongodb),实现对定时任务增、删、改、查等操作。 方法三:通过调用 flask_apscheduler 的 api (推荐) 实例对象 scheduler 拥有增、删

    2024年01月21日
    浏览(44)
  • Flask-定时任务

    目录 一、简单使用 二、apscheduler  triggers job stores schedulers 使用上下文 日志设置  Flask-APScheduler: Tips - Flask APScheduler Flask-APScheduler 是基于 python 第三方库 apscheduler 做的集成, 所以官网上只有一些简单的使用案例,详细的配置还是要看 apscheduler 的文档。 apscheduler:  Advanced Python

    2023年04月08日
    浏览(32)
  • Celery结合flask完成异步任务与定时任务

    Celery 常用于 web 异步任务、定时任务等。 使用 redis 作为 Celery的「消息代理 / 消息中间件」。 这里通过Flask-Mail使用qq邮箱延时发送邮件作为示例 使用 Flask-Mail 发送邮件需要进行一下配置,其中QQ邮箱授权码的获取方式如下所述: 点进qq邮箱,在设置里面点击账号,向下滚动开

    2024年02月07日
    浏览(53)
  • flask_apscheduler源码分析

    前言     遵循flask框架的标准的库,称为flask扩展,flask_apscheduler模块就是一个flask扩展,它使用了flask编程上下文,同时内部完全依赖apscheduler。     我近期使用flask_apscheduler遇到了一个所有job全部死亡的bug。现象:job平时是正常启动的,突然某个时刻全部挂了,所以需要分析

    2024年01月25日
    浏览(37)
  • Python自动化测试中APScheduler Flask的应用示例

    使用背景 实际项目中,需要验证打点数据在各个系统中收集是否一致,而部分节点打点数据收集是通过异步任务实现的,等待时间比较久。为应对业务异步操作处理,实现异步数据的收集,经过调研后,选择了 APScheduler 框架。 什么是 APScheduler 框架? APScheduler 是基于 Quartz

    2024年01月20日
    浏览(42)
  • STM32 实现简单定时任务调度器,动态创建任务,两种思路实现流水灯

    代码实现和硬件没关系,所以并不限于STM32,Arduino 之类的其他地方也能用,只要有一个能获取时间的函数就行,或者说,只要有一个会随着时间自动增加的变量就行,时间单位无所谓,所以确实想的话,拿到电脑上也能用。后面会用跑马灯程序来说明定时任务的玩法,可以直

    2024年02月10日
    浏览(37)
  • Cron表达式简单介绍 + Springboot定时任务的应用

    前言 表达式是一个字符串,主要分成6或7个域,但至少需要6个域组成,且每个域之间以空格符隔开。 以7个域组成的,从右往左是【年 星期 月份 日期 小时 分钟 秒钟】 秒 分 时 日 月 星期 年 以6个域组成的,从右往左是【星期 月份 日期 小时 分钟 秒钟】 秒 分 时 日 月 星

    2023年04月20日
    浏览(37)
  • 基于Mongodb分布式锁简单实现,解决定时任务并发执行问题

    我们日常开发过程,会有一些定时任务的代码来统计一些系统运行数据,但是我们应用有需要部署多个实例,传统的通过配置文件来控制定时任务是否启动又太过繁琐,而且还经常出错,导致一些异常数据的产生 网上有很多分布式锁的实现方案,基于redis、zk、等有很多,但

    2023年04月18日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包