Celery 與 Flask 大型程序結構的結合


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() 實現。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM