python自动化测试中APScheduler Flask的应用示例

以下是关于“python自动化测试中APScheduler Flask的应用示例”的详细讲解：

简介

APScheduler 是一个轻量级的 Python 定时任务调度框架，可以添加多个任务，可以动态添加修改和删除任务，先来看一下简单示例：

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    print('job start...')

scheduler = BlockingScheduler()
scheduler.add_job(job, 'interval', seconds=5)
scheduler.start()

这个简单的例子通过APScheduler实现了每五秒输出一个“job start..."的效果。

Flask是一个用Python编写的轻量级Web应用框架，具有易于扩展、关注URL和view函数的小型化等优点，也是 Python 中最流行的 Web 框架之一。

应用示例

示例一：使用APScheduler实现定时爬取数据

1. 安装APScheduler（如果没有安装）：

pip install apscheduler

1. 下载需要爬取的数据（以博客园为例）：

import requests
from bs4 import BeautifulSoup

def getData():
    url = 'https://www.cnblogs.com/'
    res = requests.get(url)
    soup = BeautifulSoup(res.text, 'html.parser')
    print(soup)

1. 配置APScheduler

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger

def timer():
    scheduler = BlockingScheduler()
    scheduler.add_job(func=getData, trigger=IntervalTrigger(seconds=30))
    scheduler.start()

if __name__ == '__main__':
    timer()

这个代码块实现了每30秒爬取一次博客园数据的效果。

1. 在Flask中实现定时任务

from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask
import datetime

app = Flask(__name__)

@app.route("/")
def hello():
    now = datetime.datetime.now()
    return f"Hello, now time is {now}"

def job():
    print('Run the job at: %s' % datetime.datetime.now())

if __name__ == '__main__':
    scheduler = BackgroundScheduler()
    scheduler.add_job(job, 'interval', seconds=10)
    scheduler.start()
    app.run()

这个代码块实现了每10秒输出“Run the job at...”的效果。

示例二：使用APScheduler和Flask实现定时任务记录

1. 配置数据库

from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

class JobRecord(db.Model):
    __tablename__ = 'jobs'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(255))
    description = db.Column(db.Text)
    execute_time = db.Column(db.DateTime)

    def __str__(self):
        return "<JobRecord(name='%s', description='%s', execute_time='%s')>" % \
               (self.name, self.description, self.execute_time)

1. 编写视图函数

from flask import jsonify
from apscheduler.schedulers.background import BackgroundScheduler
import datetime

@app.route("/")
def hello():
    now = datetime.datetime.now()
    return f"Hello, now time is {now}"

def job():
    job_record = JobRecord(name="Test Job", description="Test Job Description",
                           execute_time=datetime.datetime.now())
    db.session.add(job_record)
    db.session.commit()

@app.route("/jobs")
def joblist():
    jobs = JobRecord.query.all()
    res = []
    for j in jobs:
        res.append({
            "name" : j.name,
            "description" : j.description,
            "execute_time" : str(j.execute_time)
        })
    return jsonify(res)

if __name__ == '__main__':
    db.create_all()
    scheduler = BackgroundScheduler()
    scheduler.add_job(job, 'interval', seconds=10)
    scheduler.start()
    app.run()

在这个代码块中，我们每隔10秒钟就会往数据库中插入一条记录，同时我们还可以通过"/jobs"路由查看已经执行的所有任务次数。

这就是使用APScheduler Flask的应用示例的完整攻略，希望对你有所帮助。