Python任务调度利器之APScheduler详解

简介

APScheduler（Advanced Python Scheduler）是一个轻量级的Python任务调度库，它允许您按照指定的时间间隔或cron-like表达式调度任务执行。APScheduler是使用纯Python编写的，因此非常易于安装和使用，同时具有足够的灵活性，支持多种调度器，可以与许多后端存储系统一起使用，例如SQLAlchemy、MongoDB等。

基本使用

安装

在命令行使用pip安装APScheduler:

pip install apscheduler

简单使用

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    print('I am doing my job.')

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

上述代码会每5秒钟执行一次job函数。

调度选项

Interval调度器

interval调度器允许您按指定的秒数或分钟数执行作业。示例代码如下：

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

上述代码会每5秒钟执行一次job函数。

Cron调度器

Cron调度器是一种cron类似的作业调度器，它允许您按日期和时间执行作业。下面是cron调度器的一些示例：

# 每天的下午4点30分执行
scheduler.add_job(job, 'cron', hour=16, minute=30)

# 每月的第一天的下午4点执行
scheduler.add_job(job, 'cron', day=1, hour=16)

# 每周一的下午4点执行
scheduler.add_job(job, 'cron', day_of_week='mon', hour=16)

One-time调度器

如果您只想一次性执行作业，可以使用one-time调度器。示例代码如下：

# 5秒后执行job函数
scheduler.add_job(job, 'date', run_date='2022-12-12 12:12:12')

# 将在程序第一次运行时执行job函数
scheduler.add_job(job, 'date', run_date=datetime.now())

新版添加任务方式

新版中，使用以下方式添加任务：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstore = {
   'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}

scheduler = BackgroundScheduler(jobstores=jobstore)
scheduler.add_job(job, 'interval', seconds=5)

在该示例中，我们使用了SQLAlchemyJobStore，并将其命名为“default”。我们还创建了一个BackgroundScheduler对象，将”default"作业存储器传递给它。最后，我们按照每5秒钟执行我作业。

示例说明

示例1：使用APScheduler构建简单的邮件发送任务

任务执行时间：每天的早上8点钟

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
import smtplib
from email.mime.text import MIMEText

def send_email():
    msg = MIMEText('这是一份由APScheduler发送的测试邮件')
    msg['Subject'] = '测试邮件'
    msg['From'] = '[e-mail]'
    msg['To'] = '[e-mail]'

    smtp = smtplib.SMTP('smtp.qq.com')
    smtp.login('[e-mail]', 'password')
    smtp.sendmail('[e-mail]', '[e-mail]', msg.as_string())
    smtp.quit()
    print('邮件已发送！')

scheduler = BlockingScheduler()
scheduler.add_job(send_email, 'cron', hour=8)
scheduler.start()

示例2：使用APScheduler定时查询网站数据并存储到数据库

任务执行时间：每5分钟

from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
import requests
import sqlite3

def get_data():
    resp = requests.get('https://www.example.com/get_data')
    data = resp.json()
    # 在这里将数据写入数据库中
    conn = sqlite3.connect('data.db')
    cursor = conn.cursor()
    cursor.execute('INSERT INTO data VALUES (?, ?, ?)', (datetime.now(), data['value1'], data['value2']))   
    conn.commit()
    conn.close()

jobstore = {
   'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}

scheduler = BackgroundScheduler(jobstores=jobstore)
scheduler.add_job(get_data, 'interval', minutes=5)
scheduler.start()

在该示例中，我们每隔5分钟查询一个HTTPS网站的数据，并将其存储在SQLite数据库中。此外，也展示了使用SQLAlchemyJobStore的方式，存储任务到数据库中，以便于管理多个任务。