在 Flask 專案中使用 Celery（with 工廠模式 or not）

本文隸屬於《Flask Web 開發實戰》番外 系列。這篇文章會介紹如何在 Flask 專案中整合 Celery。

建立 Celery 程式

第一步是建立一個 Celery 程式例項。因為 Flask 程式例項通常會命名為 app，為了避免衝突，我們一般會把 Celery 例項命名為 celery 或 celery_app：

from celery import Celery

celery = Celery(__name__, broker='pyamqp://guest@localhost//')

@celery.task
def add(x, y):
return x + y

組織和載入配置

大多數教程，包括目前的 Flask 文件裡都會介紹用下面的方式載入配置：

celery.conf.update(app.config)# 這裡的 app 是 Flask 程式例項

也就是把 Celery 配置和 Flask 配置寫在一起，然後從 Flask 程式例項的配置字典裡更新配置。

但問題是，Celery 從 4.0 開始啟用新的小寫配置名，某些配置被新的名稱替換。雖然舊的大寫配置仍然支援，但如果你打算使用小寫配置名，或是打算在未來進行遷移，這裡的配置載入方式就會失效，因為 Flask 在從檔案或物件匯入配置時只會匯入大寫形式的配置變數。

因此，我建議將 Celery 配置寫在單獨的檔案裡，不要和 Flask 配置混在一起。按照 Celery 文件的示例，你可以在當前目錄建立一個celeryconfig.py 檔案（或是其他名字）儲存配置：

broker_url = 'pyamqp://'
result_backend = 'rpc://'

task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Europe/Oslo'
enable_utc = True

然後使用下面的方法載入配置（使用其他模組名，或是在其他路徑時，記得修改傳入的字串）：

celery.config_from_object('celeryconfig')

如果需要在建立 Celery 例項時傳入 broker 和 backend 配置，可以直接寫出或是從配置模組中匯入：

from celeryconfig import broker_url

celery = Celery(__name__, broker=broker_url)

在 Flask 程式中初始化 Celery

你可以單獨建立 Celery 程式，但我們通常會需要為它新增 Flask 程式上下文支援，因為有時候你的 Celery 任務函式會用到依賴於 Flask 程式上下文的某些變數。

下面我們為 Celery 建立了一個工廠函式，在工廠函式中建立 Celery 例項，載入配置，並實現 Flask 程式上下文支援：

from flask import Flask
from celery import Celery

from celeryconfig import broker_url


def make_celery(app):
celery = Celery(__name__, broker=broker_url)
celery.config_from_object('celeryconfig')

class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)

celery.Task = ContextTask
return celery

app = Flask(__name__)

celery = make_celery(app)

在定義 Celery 任務的模組裡，比如 tasks.py，你可以匯入這個 Celery 程式例項：

from app import celery

@celery.task
def add(x, y):
return x + y

在使用工廠函式的 Flask 程式中初始化 Celery

當 Flask 程式也使用工廠函式建立時，我們可以全域性建立 Celery 程式例項，然後在建立 Flask 程式例項的工廠函式裡更新 Celery 程式配置並進行上下文設定：

from flask import Flask
from celery import Celery

from celeryconfig import broker_url
 
celery = Celery(__name__, broker=broker_url)
 

def create_app():
app = Flask(__name__)
 
register_celery(app)
return app
 

def register_celery(app):
celery.config_from_object('celeryconfig')
 
class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
 
celery.Task = ContextTask

同樣直接匯入 Celery 例項並建立任務：

from app import celery

@celery.task
def add(x, y):
return x + y

這本來是一個完整的 Celery 入門教程，但因為去年的一次硬碟損壞，對應的示例程式弄丟了，暫時沒有毅力重寫一遍，所以這篇文章只抽取了其中和 Flask 相關的內容。

因為距離初稿寫作的時間已經過去半年多，Celery 的最新版本也已經是 4.3.0，如果文中有什麼疏漏，或是有更好的實現方式，歡迎評論指出。