基于Fastapi实现定时任务

简介

定时任务是一个通用场景常见的功能，之前我使用django的时候，更习惯使用celery中的定时任务，现在花时间看了看apscheduler感觉不错，就写了demo，并集成到项目代码中了

任务调度主要就是以下几个功能

• 添加/删除 任务调度

• 暂停/恢复 任务调度(这条我未实现)

• 查看定时任务状态

实现

添加定时任务

其中添加定时任务方式，有以下三种方式

• date: 固定的时间执行一次时 用这种

• interval: 想要在固定的间隔时间循环执行时用这种

• cron: 这种就是最为灵活的 crontab 表达式定时任务了

  
  

在FastAPI异步框架中，选择 AsyncIOScheduler调度程序

默认使用sqlite持久化定时任务，不至于重启就失效

from apscheduler.schedulers.asyncio import AsyncIOSchedulerfrom apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStorefrom apscheduler.triggers.cron import CronTrigger
Schedule = AsyncIOScheduler(    jobstores={        'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')    })Schedule.start()

不完善的地方

• 1 由于时间仓促，比如参数验证的部分，参数数据校验，还可以继续完善。

• 2 添加调度任务参数封装，我把三种添加任务的三种方式，拆成了三个函数，其中很多参数没有用到。

• 3 其中文档中还有说明最大woker数量限制之类的，然后按照我使用其他定时任务的常识，应该还会函数最长执行时间限制(比如执行的功能函数特别耗时)

• 4 ...等等

  
  

这些就需要自己查看文档和issues搜索了

其他方案

定时任务有很多种方案，比如可以使用

• arq https://github.com/samuelcolvin/arq 目前看来 star数量不多

• fastapi-utils https://github.com/dmontagu/fastapi-utils

• celery https://github.com/celery/celery Celery就是Python中最为重量级，且常见的任务队列， 你也可以只用其中的定时任务功能。

说到arq, 想起我之前，使用过rq 并且学习的时候稍微翻译了一下文档 rq v1.0 https://codercharm.github.io/Python-rq-doc-cn/#/ 一晃过去一年了。

代码地址

• 单文件例子: https://github.com/CoderCharm/fastapi-mysql-generator/blob/master/examples/demo_scheduler/main.py

• 项目中使用: https://github.com/CoderCharm/fastapi-mysql-generator/blob/master/{{cookiecutter.project_name}}/app/api/__init__.py#L230

参考地址

• apscheduler官方文档

• FastAPI issues

原文来自

https://www.cnblogs.com/CharmCode/p/14191009.html

10T 技术资源大放送！包括但不限于：Linux、虚拟化、容器、云计算、网络、Python、Go 等。在 开源Linux 公众号内回复 10T，即可免费获取！
Linux学习指南
有收获，点个在看