[[439191]]

FastAPI 是 Python Web 領(lǐng)域非常受歡迎的框架，目前 GitHub 上有 39.1k 的 star，已經(jīng)遠(yuǎn)超過了 Django rest framework(22.3k star)。而 Celey 又是異步任務(wù)最流行的框架，常用于數(shù)據(jù)挖掘和機器學(xué)習(xí)等計算密集型任務(wù)的場景中。如果需要通過 API 來異步調(diào)用任務(wù)，那這兩個框架可以放在一起工作。本文來分享一下如何讓 FastAPI 和 Celery 更好的相互配合，開發(fā)環(huán)境下如何通過一個命令就可以讓兩者一起工作。

0、安裝依賴

pip install fastapi celery uvicorn 

1、寫個純 celery 任務(wù)

首先，讓我們來寫一個純屬 celery 的任務(wù)，讓它正常運行，然后在通過 fastapi 來調(diào)用它。

假設(shè)你的機器已經(jīng)安裝了 Python3 和 celery，并且本機已經(jīng)開啟了 redis，運行在 6379 端口上。

現(xiàn)在讓我們寫一個簡單的任務(wù)：計算兩數(shù)之和，文件名為：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 接收遠(yuǎn)程調(diào)用。

celery -A celery_app worker -l info 

如果要遠(yuǎn)程異步調(diào)用這個 add 函數(shù)，我們需要再編寫一個文件 start_celery_app.py，內(nèi)容如下：

from celery_app import add #導(dǎo)入我們的任務(wù)函數(shù)add 
import time 
result = add.delay(12,12) #異步調(diào)用，這一步不會阻塞，程序會立即往下運行 
 
while not result.ready():# 循環(huán)檢查任務(wù)是否執(zhí)行完畢 
    print(time.strftime("%H:%M:%S")) 
    time.sleep(1) 
 
print(result.get()) #獲取任務(wù)的返回結(jié)果 
print(result.successful()) #判斷任務(wù)是否成功執(zhí)行 

任務(wù)返回了結(jié)果 24，命令成功完成，

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

2、通過 fastapi 來執(zhí)行

編寫一個 api.py 通過接口來調(diào)用上述的 add 函數(shù)：

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()} 

啟動服務(wù)：

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

然后訪問：http://127.0.0.1:8000，會發(fā)現(xiàn)任務(wù)成功執(zhí)行：

這樣我們啟動了兩個獨立的進程，一個是 celery 的 worker，一個是 fastapi 的 app，這樣做沒問題，且生產(chǎn)環(huán)境下是嚴(yán)格要求分開運行的，不過，開發(fā)環(huán)境下如果這樣做就太低效了。

3、開發(fā)環(huán)境下如何一條命令啟動

如果不使用兩個終端來啟動兩個命令，我們可以使用 Celery 提供的測試實用程序在后臺線程中啟動 celery worker，比如寫一個這樣的文件run.py，內(nèi)容如下：

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() 

這樣，只需要執(zhí)行一條命令就可以同時啟動 celery worker 和 fastapi 接口服務(wù)，調(diào)試的時候是不是非常方便：圖片

最后的話

本文分享了 fastapi 和 celery 是如何配合工作的，并分享了一個用于開發(fā)環(huán)境的腳本，可以通過一個命令來啟動 celery worker 和 fastapi，可能不是完美的解決方案，但確實提升了開發(fā)效率，我覺得這是值得的，如果有幫助還請點贊、在看、關(guān)注，感謝閱讀。

本文轉(zhuǎn)載自微信公眾號「Python七號」，可以通過以下二維碼關(guān)注。轉(zhuǎn)載本文請聯(lián)系Python七號公眾號。