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 为空,因此调取上下文将引发"RunTimeError"的报错。
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() 实现。
原文地址:https://www.cnblogs.com/autopenguin/p/6542899.html