Flask-APScheduler 定时运行api接口

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS PostgreSQL,集群系列 2核4GB
简介: Flask-APScheduler 定时运行api接口

1.APScheduler简介


APScheduler是一个Python库,可让您安排稍后执行的Python代码,是一套任务调度框架,可以用来做定时任务控制器,可以添加删除任务。如果将作业存储在数据库中,它们也将在调度程序重新启动后继续运行并保持其状态。重新启动调度程序后,它将运行它应该在脱机时运行的所有作业。


😁我这里先给出一个写调度api的模版吧


步骤一:写好自己运行的函数 def fuction()

步骤二:创建flask路由,在路由下定义一个需要调度的函数def job_function():,把之前的def fuction()函数放到里面

步骤三:创建调度任务函数,def task(),这里需要配置任务调度的方式等参数

步骤四:运行任务 task(),主要写在主函数外面

步骤五:主函数运行api接口,app.run()


from flask import Flask
from flask_apscheduler import APScheduler
import datetime
def fuction()
    print("showmaker")
app = Flask(__name__)
@app.route("/test", methods=['GET','POST'])
def job_function():
    fuction()
def task():
    scheduler = APScheduler()
    scheduler.init_app(app)
    # 定时任务的格式
    scheduler.add_job(func=job_function, trigger='interval',seconds=3600, id='my_job_id')
    scheduler.start()
# 写在main里面,IIS不会运行
task()
if __name__ == "__main__":
    app.run(debug=False, port="5005")

2.APScheduler调度方式


APScheduler有三种可以调度任务的方式:


我们的目的就是通过api接口,在终端一直运行这个接口进程,让它自己去调度我们的程序,这里的程序也就是任务。


  1. 可以选择调度的任务,开始和结束的时间
  2. 基于一个时间区间,间隔的执行任务
  3. 一次性任务执行


3.APScheduler组件


知道了我们的任务有哪些调度方式,现在了解一下APScheduler的组件:


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

3.1 triggers


triggers: 支持三种任务触发方式

date:


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


参数 说明
run_date 任务运行日期
timezone 时区


scheduler.add_job(func=job_function, trigger='date',run_date='2021-10-27 13:00:00', id='my_job_id')


interval:

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

参数 说明
weeks (int) 间隔几周
days (int) 间隔几天
hours (int) 间隔几小时
minutes (int) 间隔几分钟
seconds (int) 间隔多少秒
start_date (datetime 或 str) 开始日期
end_date (datetime 或 str) 结束日期


scheduler.add_job(func=job_function, trigger='interval',seconds=3600, id='my_job_id')

cron

cron风格的任务触发,这个不常用就不介绍了,我懒


3.2 job stores:支持四种任务存储方式


memory:默认配置任务存在内存中

mongdb:支持文档数据库存储

sqlalchemy:支持关系数据库存储

redis:支持键值对数据库存储

这里如果你的数据库是mysql,我推荐使用sqlalchemy作为数据库,用过create_engine来连接数据库实现数据库操作,比如说删除数据记录,写入数据记录。这里是我自己的例子


from flask import Flask
from flask_apscheduler import APScheduler
import datetime
import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy import MetaData,Table
def delete():
    engine = create_engine('mysql+pymysql://用户名:密码@ip:端口号/数据库?charset=gbk')
    meta = MetaData(bind=engine)
    tb_1 = Table('数据表名', meta, autoload=True, autoload_with=engine)
    tb_2 = Table('数据表名', meta, autoload=True, autoload_with=engine)
    dlt_1=tb_1.delete()
    dlt_2=tb_2.delete()
    conn = engine.connect()
    # 执行delete操作
    conn.execute(dlt_1)
    conn.execute(dlt_2)
    conn.close()
def write(t):
    r1=pd.DataFrame()
    r3=pd.DataFrame()
    engine = create_engine('mysql+pymysql://用户名:密码@ip:端口号/数据库?charset=gbk')
    r1.to_sql('数据表名', con=engine, if_exists='append', index=False)
    r3.to_sql('数据表名', con=engine, if_exists='append', index=False)
def write_log(buf):
    print(buf)
    with open('test.txt', 'a') as f:
        f.write(buf + "\n")
app = Flask(__name__)
@app.route("/test", methods=['GET','POST'])
def job_function():
    now_time=datetime.datetime.now()
    t=now_time.strftime('%Y-%m-%d')
    write_log("update time:"+t)
    delete()
    write_log("delete data")
    write(t)
    write_log("write data")
def task():
    scheduler = APScheduler()
    scheduler.init_app(app)
    # 定时任务,每隔10s执行1次
    scheduler.add_job(func=job_function, trigger='interval',seconds=60, id='my_job_id')
    scheduler.start()
# 写在main里面,IIS不会运行
task()
if __name__ == "__main__":
    app.run(debug=False, port="8888")


最后可以在test.txt查看自己的调用的时间


3.3 schedulers:


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


BlockingScheduler: 当这个调度器是你应用中 唯一要运行 的东西时使用

BackgroundScheduler: 当不运行其它框架 的时候使用,并使你的任务在 后台运行

AsyncIOScheduler: 当你的程序是 异步IO模型 的时候使用

GeventScheduler: 和 gevent 框架配套使用

TornadoScheduler: 和 tornado 框架配套使用

TwistedScheduler: 和 Twisted 框架配套使用

QtScheduler: 开发 qt 应用的时候使用

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
相关文章
|
11天前
|
JSON 关系型数据库 测试技术
使用Python和Flask构建RESTful API服务
使用Python和Flask构建RESTful API服务
|
16天前
|
JSON API 数据格式
淘宝 / 天猫官方商品 / 订单订单 API 接口丨商品上传接口对接步骤
要对接淘宝/天猫官方商品或订单API,需先注册淘宝开放平台账号,创建应用获取App Key和App Secret。之后,详细阅读API文档,了解接口功能及权限要求,编写认证、构建请求、发送请求和处理响应的代码。最后,在沙箱环境中测试与调试,确保API调用的正确性和稳定性。
|
29天前
|
供应链 数据挖掘 API
电商API接口介绍——sku接口概述
商品SKU(Stock Keeping Unit)接口是电商API接口中的一种,专门用于获取商品的SKU信息。SKU是库存量单位,用于区分同一商品的不同规格、颜色、尺寸等属性。通过商品SKU接口,开发者可以获取商品的SKU列表、SKU属性、库存数量等详细信息。
|
1月前
|
JSON API 数据格式
如何使用Python和Flask构建一个简单的RESTful API。Flask是一个轻量级的Web框架
本文介绍了如何使用Python和Flask构建一个简单的RESTful API。Flask是一个轻量级的Web框架,适合小型项目和微服务。文章从环境准备、创建基本Flask应用、定义资源和路由、请求和响应处理、错误处理等方面进行了详细说明,并提供了示例代码。通过这些步骤,读者可以快速上手构建自己的RESTful API。
33 2
|
1月前
|
JSON API 数据格式
店铺所有商品列表接口json数据格式示例(API接口)
当然,以下是一个示例的JSON数据格式,用于表示一个店铺所有商品列表的API接口响应
|
2月前
|
编解码 监控 API
直播源怎么调用api接口
调用直播源的API接口涉及开通服务、添加域名、获取API密钥、调用API接口、生成推流和拉流地址、配置直播源、开始直播、监控管理及停止直播等步骤。不同云服务平台的具体操作略有差异,但整体流程简单易懂。
|
20天前
|
JSON API 数据安全/隐私保护
拍立淘按图搜索API接口返回数据的JSON格式示例
拍立淘按图搜索API接口允许用户通过上传图片来搜索相似的商品,该接口返回的通常是一个JSON格式的响应,其中包含了与上传图片相似的商品信息。以下是一个基于淘宝平台的拍立淘按图搜索API接口返回数据的JSON格式示例,同时提供对其关键字段的解释
|
2月前
|
JSON API 数据格式
使用Python和Flask构建简单的RESTful API
【10月更文挑战第12天】使用Python和Flask构建简单的RESTful API
47 1
|
2月前
|
API 数据库 网络架构
深入浅出:使用Python Flask实现RESTful API
【10月更文挑战第7天】在数字化时代,掌握如何高效构建和部署RESTful API是后端开发者的必备技能。本文将引导你了解如何使用Python Flask框架快速打造一个简单而强大的RESTful服务。从基础环境搭建到API设计原则,再到实际代码示例,我们将一步步揭开Flask框架的神秘面纱,让你轻松上手,并能够自信地处理更复杂的项目。
|
2月前
|
人工智能 自然语言处理 PyTorch
Text2Video Huggingface Pipeline 文生视频接口和文生视频论文API
文生视频是AI领域热点,很多文生视频的大模型都是基于 Huggingface的 diffusers的text to video的pipeline来开发。国内外也有非常多的优秀产品如Runway AI、Pika AI 、可灵King AI、通义千问、智谱的文生视频模型等等。为了方便调用,这篇博客也尝试了使用 PyPI的text2video的python库的Wrapper类进行调用,下面会给大家介绍一下Huggingface Text to Video Pipeline的调用方式以及使用通用的text2video的python库调用方式。