开发环境下，如何通过一个命令让 Fastapi 和 Celery 一起工作

FastAPI 是环境i和 Python Web 领域非常受欢迎的框架，目前 GitHub 上有 39.1k 的下何 star，已经远超过了 Django rest framework(22.3k star)。通过而 Celey 又是个命异步任务最流行的框架，常用于数据挖掘和机器学习等计算密集型任务的令让场景中。如果需要通过 API 来异步调用任务，起工那这两个框架可以放在一起工作。环境i和本文来分享一下如何让 FastAPI 和 Celery 更好的下何相互配合，开发环境下如何通过一个命令就可以让两者一起工作。通过

0、个命安装依赖

pip install fastapi celery uvicorn 

1、令让写个纯 celery 任务

首先，起工让我们来写一个纯属 celery 的环境i和任务，让它正常运行，下何然后在通过 fastapi 来调用它。通过

假设你的机器已经安装了 Python3 和 celery，并且本机已经开启了 redis，运行在 6379 端口上。

现在让我们写一个简单的任务：计算两数之和，文件名为：celery_app.py 代码如下：

#!/Users/aaron/py38env/bin/python # filename: celery_app.py from celery import Celery app = Celery("tasks", broker=redis://127.0.0.1:6379/0, backend=redis://127.0.0.1:6379/0) @app.task def add(x, y):     return x + y 

然后启动一个 worker 接收远程调用。

celery -A celery_app worker -l info 

如果要远程异步调用这个 add 函数，源码下载我们需要再编写一个文件 start_celery_app.py，内容如下：

from celery_app import add #导入我们的任务函数add import time result = add.delay(12,12) #异步调用，这一步不会阻塞，程序会立即往下运行 while not result.ready():# 循环检查任务是否执行完毕     print(time.strftime("%H:%M:%S"))     time.sleep(1) print(result.get()) #获取任务的返回结果 print(result.successful()) #判断任务是否成功执行 

任务返回了结果 24，命令成功完成，

此时 worker 界面增加的信息如下：

2、通过 fastapi 来执行

编写一个 api.py 通过接口来调用上述的 add 函数：

from fastapi import FastAPI import celery_app app = FastAPI() @app.get("/") def read_root():     result = celery_app.add.delay(12, 12)     return { "12+12": result.get()} 

启动服务：

uvicorn api:app --host 0.0.0.0 --port 8000 --reload 

然后访问：http://127.0.0.1:8000，会发现任务成功执行：

这样我们启动了两个独立的进程，一个是 celery 的 worker，一个是 fastapi 的 app，这样做没问题，且生产环境下是严格要求分开运行的，不过，开发环境下如果这样做就太低效了。

3、开发环境下如何一条命令启动

如果不使用两个终端来启动两个命令，我们可以使用 Celery 提供的网站模板测试实用程序在后台线程中启动 celery worker，比如写一个这样的文件run.py，内容如下：

import uvicorn original_callback = uvicorn.main.callback def callback(**kwargs):     from celery.contrib.testing.worker import start_worker     from celery_app import app     with start_worker(app, perform_ping_check=False, loglevel="info"):         original_callback(**kwargs) uvicorn.main.callback = callback if __name__ == "__main__":     uvicorn.main() 

这样，只需要执行一条命令就可以同时启动 celery worker 和 fastapi 接口服务，调试的时候是不是非常方便：图片

最后的话

本文分享了 fastapi 和 celery 是如何配合工作的，并分享了一个用于开发环境的脚本，可以通过一个命令来启动 celery worker 和 fastapi，可能不是完美的解决方案，但确实提升了开发效率，我觉得这是值得的，如果有帮助还请点赞、在看、关注，感谢阅读。

本文转载自微信公众号「Python七号」，可以通过以下二维码关注。转载本文请联系Python七号公众号。

云服务器

很赞哦!（5488）