在 Flask 应用中实现定时任务可以帮助我们自动化执行一些周期性的任务,比如数据备份、发送邮件通知等。以下是一个简单的示例教程,展示了如何使用 Flask 结合 apscheduler 库来实现定时任务。
准备工作
首先,确保你已经安装了 Flask 和 apscheduler。如果没有安装,可以通过以下命令进行安装:
pip install Flask apscheduler
创建 Flask 应用
- 初始化 Flask 应用:
from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
app = Flask(__name__)
# 创建一个调度器实例
scheduler = BackgroundScheduler()
# 启动调度器
scheduler.start()
# 定义一个定时任务
def scheduled_task():
print("定时任务执行中...")
# 添加定时任务,每天凌晨 1 点执行
scheduler.add_job(scheduled_task, 'cron', hour=1)
if __name__ == '__main__':
app.run(debug=True)
- 解释代码:
BackgroundScheduler():创建一个后台调度器实例。scheduler.start():启动调度器。scheduled_task():定义一个要执行的函数。scheduler.add_job(scheduled_task, 'cron', hour=1):添加一个定时任务,使用 cron 表达式来指定任务执行的时间。在这个例子中,任务每天凌晨 1 点执行。
运行 Flask 应用
将上面的代码保存为一个 Python 文件,例如 app.py。然后在命令行中运行以下命令来启动 Flask 应用:
python app.py
现在,你的 Flask 应用应该已经启动,并且每天凌晨 1 点会自动执行 scheduled_task 函数。
扩展功能
- 添加更多的定时任务:
scheduler.add_job(scheduled_task, 'cron', hour=2, minute=30)
这个例子中,我们添加了一个新的定时任务,每天凌晨 2 点 30 分执行。
- 使用不同的触发器:
apscheduler 支持多种触发器,如 date、interval 等。以下是一个使用 interval 触发器的例子:
scheduler.add_job(scheduled_task, 'interval', seconds=60)
这个例子中,任务每 60 秒执行一次。
- 使用装饰器:
你可以使用 @app.route 装饰器来创建一个 Flask 路由,用于手动触发定时任务:
@app.route('/run-task')
def run_task():
scheduled_task()
return "任务已执行"
现在,当你访问 /run-task 路由时,scheduled_task 函数将被执行。
通过以上示例,你可以了解到如何使用 Flask 和 apscheduler 实现简单的定时任务。在实际应用中,你可以根据需求调整定时任务的时间、触发器和执行函数。
