在flask微服务中使用调度器设置定时任务：BackgroundScheduler ...

在flask微服务中使用调度器设置定时任务：BackgroundScheduler

1. #!/usr/bin/env python3
2. # -*- coding: UTF-8 -*-
3. # filename: main.py
5. import os
6. import unittest
8. from app import create_app, db
9. from app.route import route
10. from app.service.score_service import send_score_mail
11. from apscheduler.schedulers.background import BackgroundScheduler
14. # 环境变量
15. env = os.getenv('RUNTIME_ENV') or 'dev'
17. # 创建应用
18. app = create_app(env)
20. # 服务路由
21. route(app)
23. def start_scheduler():
24.     # 创建调度器
25.     scheduler = BackgroundScheduler(timezone='Asia/Shanghai')
26.     # 添加定时任务
27.     scheduler.add_job(send_score_mail, 'cron', hour=7, minute=0, day_of_week='mon')
28.     # 启动调度器
29.     scheduler.start()
30.     return scheduler
33. # 启动调度器（放在一个函数中，但在这里仍然调用它以保持现有行为）
34. _scheduler = start_scheduler()
37. @app.cli.command()
38. def create():
39.     """ Create all database. """
40.     db.drop_all()
41.     db.create_all()
42.     db.session.commit()
45. @app.cli.command()
46. def test():
47.     """ Run service unittest. """
48.     ut = unittest.TextTestRunner()
49.     ut.run(unittest.TestLoader().discover('app/test'))
53. if __name__ == '__main__':
54.     app.run(debug=False, host='0.0.0.0', port=5003) # 此处debug为False时定时任务只会执行一次，debug为True时会执行2次

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。