Python定时任务工具Flask-APScheduler基本功能：作业的新增、起、停介绍

一直想找一个定时调度的工具，前几天忽然发现Python有一个APScheduler，官方的描述（http://apscheduler.readthedocs.io/en/latest/）是这样的：

Advanced Python Scheduler (APScheduler) is a Python library that lets you schedule your Python code to be executed later, either just once or periodically.

然后发现了在这上面有一个flask-apscheduler，提供了基本的服务接口。

我觉得很符合我的要求，后期和Celery结合起来分布式调度框架也就有了。

开始先测试一下简单点功能，

验证计划如下：

步骤1：

1. 能够增加新的定时
   
2. 能够暂停/开始已有的作业

步骤2：

1. 1.任务信息存储到数据库中

步骤3：

1. 查看所有job列表
2. 查看job具体信息
3. 测试一些实际案例

步骤4：

1. 和Celery集成完成初步跨机器调度

步骤1具体说明：

0.安装

 pip install Flask-APScheduler

 pip install sqlalchemy  #由于计划使用pg作为存储库，先安装。

1.测试代码简介

在examples/jobs.py的基础上修改了下，提供了增加job、暂停和恢复3个功能

核心代码如下：

def job1(a, b):

    print(str(a) + ' ' + str(b))

def jobfromparm(**jobargs):

    id = jobargs['id']

    func = jobargs['func']

    args = eval(jobargs['args'])

    trigger = jobargs['trigger']

    seconds = jobargs['seconds']

    print('add job: ',id)

    job = scheduler.add_job(func=func,id=id, args=args,trigger=trigger,seconds=seconds)

    return 'sucess'

@app.route('/pause')

def pausejob():

    scheduler.pause_job('job1')

    return "Success!"

@app.route('/resume')

def resumejob():

    scheduler.resume_job('job1')

    return "Success!"

@app.route('/addjob', methods=['GET', 'POST'])

def addjob():

    data = request.get_json(force=True)

    job = jobfromparm(**data)

2.测试结果

使用curl测试即可，注意其中参数内容可以参考apscheduler的add_job函数，如果要增加需要确保id的修改

curl -i -X POST -H "'Content-type':'appon/x-www-form-urlencoded', 'charset':'utf-8', 'Accept': 'text/plain'" -d '{"id":"job1","func": "jobs:job1","args":"(1, 8)","trigger":"interval","seconds":10}' http://127.0.0.1:5000/addjob



curl -i -X POST -H "'Content-type':'appon/x-www-form-urlencoded', 'charset':'utf-8', 'Accept': 'text/plain'" -d '{"id":"job2","func": "jobs:job1","args":"(2, 8)","trigger":"interval","seconds":10}' http://127.0.0.1:5000/addjob

示例结果：

附件是具体代码：

• test.py