FastAPI 中怎么设置定时任务？FastAPI 中设置定时任务的方法

在 FastAPI 中设置定时任务，可以使用 APScheduler、celery 或者 asyncio 等模块。APScheduler 使用简单，asyncio 适合与 FastAPI 的异步特性。

用 Apifox，节省研发团队的每一分钟

FastAPI 中怎么设置定时任务？FastAPI 中设置定时任务的方法

免费使用 Apifox

相关推荐

最新文章

API

一体化协作平台

API 设计

API 文档

API 调试

自动化测试

API Mock

API Hub

立即体验 Apifox
目录

在 Web 应用中，我们经常需要执行一些定时任务，比如定时抓取数据、定时任务调度等。在 FastAPI 中，我们可以通过以下几种方法来实现定时任务：

  • 使用 APScheduler
  • 使用后台任务队列 celery
  • 使用 asyncio 的定时任务
FastAPI 中设置定时任务的方法
FastAPI 官网

使用 APScheduler

APScheduler 是 Python 中一个非常常用的定时任务框架。要在 FastAPI 中使用它，主要分以下几步：

1.安装 APScheduler:

pip install APScheduler


2.在 FastAPI 应用中导入 APScheduler 并初始化：

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()


3.定义定时任务方法：

def tick():
    print('Tick! The time is: %s' % datetime.now())


4.在 FastAPI 应用启动完成后，启动 APScheduler:

@app.on_event("startup")  
async def startup_event():
    scheduler.add_job(tick, 'interval', seconds=3)
    scheduler.start()

这样就可以每 3 秒执行一次 tick 方法。我们也可以通过 scheduler 的其他方法来设置更复杂的定时任务。

使用 celery

celery 也是一个非常常用的任务队列，可以配合 FastAPI 使用。
1.安装 celery:

pip install celery


2.定义 celery app 和任务：

from celery import Celery

celery_app = Celery(__name__)

@celery_app.task  
def celery_task():
    print('celery task done')


3.在 FastAPI 中导入 celery app 并设置定时任务：

from celery.schedules import crontab  

@app.on_event("startup")
async def startup():
    celery_app.conf.beat_schedule = {
        'beat-task-every-30-seconds': {
            'task': 'celery_task',
            'schedule': 30.0, 
        },
    }

这样就可以每 30 秒执行一次 celery 任务。

使用 asyncio

Python 中的 asyncio 模块也可以用来实现定时任务。

import asyncio

@app.on_event("startup")
async def startup_event():
    asyncio.create_task(run_tasks()) 

async def run_tasks():
    while True:
        print('Cron job running') 
        await asyncio.sleep(10)

上面代码会每 10 秒执行一次定时任务。

实践案例

这里给出一个使用 APScheduler 的完整示例：

from fastapi import FastAPI
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler

# 初始化FastAPI实例
app = FastAPI() 

# 初始化后台调度器
scheduler = BackgroundScheduler()

# 定时任务函数
def tick():
    print('Tick! The time is: %s' % datetime.now())

# FastAPI启动完成事件
@app.on_event("startup")  
async def startup_event():
    # 添加定时任务,间隔3秒钟执行tick函数
    scheduler.add_job(tick, 'interval', seconds=3)
    # 启动调度器
    scheduler.start()
    
# 基本路由    
@app.get("/")
async def root():
    return {"message": "Hello World"}
    
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

这个示例会在启动时开始一个后台定时任务，每 3 秒执行一次 tick 方法。

FastAPI 定时时间实践案例结果
FastAPI 定时时间实践案例结果

提示、技巧和注意事项

  • 使用 Celery 时，确保已安装和配置 Redis 作为消息队列。
  • 在设置定时任务时，要小心任务的执行时间，以免对应用性能产生不利影响。
  • 可以使用try...except块来捕获定时任务中的异常，并进行适当的错误处理

使用 Apifox 调试 FastAPI 接口

如果你是 FastAPI 开发者，你经常需要与 API 打交道，确保你的应用程序能够正常工作。这时，一个强大的接口测试工具就会派上用场。

Apifox 是一个比 Postman 更强大的接口测试工具，Apifox = Postman + Swagger + Mock + JMeter，Apifox 支持调试 http(s)、WebSocket、Socket、gRPC、Dubbo 等协议的接口，并且集成了 IDEA 插件。在写完服务接口时，测试阶段可以通过 Apifox 来校验接口的正确性，图形化界面极大的方便了项目的上线效率。


如果想快速的调试一条接口，新建一个项目后，在项目中选择“调试模式”，填写请求地址后即可快速发送请求，并获得响应结果，上文的实践案例如图所示：

立即体验 Apifox
Apifox 调试后端 FastAPI 接口
Apifox 调试后端接口

总结

在 FastAPI 中设置定时任务，可以使用 APScheduler、celery 或者 asyncio 等模块。APScheduler 使用简单，asyncio 适合与 FastAPI 的异步特性。可以根据实际场景及需求选择不同的方法。

知识扩展：


参考链接：

  • APScheduler Documentation: https://apscheduler.readthedocs.io/en/latest/index.html
  • Celery Documentation: https://docs.celeryq.dev/en/stable/index.html
  • asyncio Documentation: https://docs.python.org/3/library/asyncio.html
fastapi Python
相关推荐