Celery 與 Flask 大型程序結構結合的關鍵:
使 Celery 與 Flask 結合的關鍵在於非 Web Runtime 中調取上下文。
因為當 Flask App 在作為 WSGI Application 運行時,
會在每個請求進入的時候將請求上下文推入
_request_ctx_stack
中。推入請求上下文的同時,將檢查應用上下文堆棧
_app_ctx_stack
是否
為空,若空則
隱式
地推入一個 應用上下文。
最終,在請求線程退出前,應用上下文將從其
Flask._app_ctx_stack
的棧中
里彈出。
Celery 作為任務調度模塊,其任務執行單元(worker)運行時,將處於非 Web Runtime 的狀態,而此時應用上下文堆棧
Flask._app_ctx_stack
為空,因此調取上下文將引發"RunTime
Error
"的報錯。
Flask 大型程序結構下,采用工廠函數(create_app)創建 Flask App,因此同樣采用工廠函數的形式創建 Celery App,其工廠函數如下:
from celery import Celery
def make_celery(App):
CeleryInst = Celery(App.import_name, backend = App.config['CELERY_RESULT_BACKEND'], broker = App.config['CELERY_BROKER_URL'])
CeleryInst.conf.update(App.config)
TaskBase = CeleryInst.Task
class ContextTask(TaskBase):
"""Will be execute when create the instance object of ContextTasks."""
# Will context(Flask's Extends) of app object(Producer Sit)
# be included in celery object(Consumer Site).
abstract = True
def __call__(self, *args, **kwargs):
with App.app_context():
return TaskBase.__call__(self, *args, **kwargs)
# Include the app_context into celery.Task.
# Let other Flask extensions can be normal calls.
CeleryInst.Task = ContextTask
return CeleryInst
注意:此方式創建 Celery 對象時,將會把用於初始化的 Flask App 的上
下文信息包含至 Celery 對象中。基於此,Flask 擴展在調用 Celery 任務時,采用最普通的方式,而不需額外考慮上下文問題。
Celery 與 Flask 大型程序結構結合的實例:
Flask 大型程序結構如下。
|-flasky
|-app/
|-templates/
|-static/
|-main/
|-__init__.py
|-errors.py
|-forms.py
|-views.py
|-__init__.py
|-mail/
|-templates/
|-__init__.py
|-mail.py
|-migrations/
|-tests/
|-__init__.py
|-test*.py
|-venv/
|-requirements.txt
|-config.py
|-manage.py
注意:與原 Flask 大型程序結構作出如下
修改以及強調。
(1).Flask App 及 Celery App 創建於 /flasky/app/__init__.py 中。
(2).manage.py 中通過 import 方式引入 Flask App,並在該文件中啟用 WSGI。
(3).在其他需要調用 Flask App 的地方,采用
current_app._get_current_object() 實現。