devbox
凡人多烦事01
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

在Celery中使用Flask的上下文

作者：凡人多烦事01 | 2024-04-13 04:00:01

赞

踩

如何在celery异步任务中引入 flask的应用上下文

其实我只是想把邮件发送这个动作移到Celery中执行。
既然用到了Celery，那么每次发邮件都单独开一个线程似乎有点多余，异步任务还是交给Celery吧。

在Flask应用中集成Celery

Celery和Flask一起使用并没有什么不和谐的地方，都可以不用定制的Flask扩展，按照网上随处可见的示例也很简单：

  1. from flask import Flask
  2. from celery import Celery
  3.  
  4. app = Flask(__name__)
  5. app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
  6. app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
  7.  
  8. celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
  9. celery.conf.update(app.config)
  10.  
  11. @celery.task
  12. def send_email():
  13.     ....

然而，稍微上点规模的Flask应用都会使用Factory模式（中文叫工厂函数，我听着特别扭），即只有在创建Flask实例时，才会初始化各种扩展，这样可以动态的修改扩展程序的配置。比如你有一套线上部署的配置和一套本地开发测试的配置，希望通过不同的启动入口，就使用不同的配置。
使用Factory模式的话，上面的代码大概要修改成这个样：

  1. from flask import Flask
  2. from celery import Celery
  3.  
  4. app = Flask(__name__)
  5. celery = Celery()
  6.  
  7. def create_app(config_name):
  8.     app.config.from_object(config[config_name])
  9.     celery.conf.update(app.config)

通过config_name，来动态调整celery的配置。然而，这样子是不行的！
Celery的__init__()函数会调用celery._state._register_app()直接就通过传入的配置生成了Celery实例，上面的代码中，celery = Celery()直接使用默认的amqp作为了broker，随后通过celery.conf.update(app.config)是更改不了broker的。这也就是为什么网上的示例代码中，在定义Celery实例时，就传入了broker=app.config['CELERY_BROKER_URL']，而不是之后通过celery.conf.update(app.config)传入。当你的多套配置文件中，broker设置的不同时，就悲剧了。

当然不用自己造轮子，Flask-Celery-Helper就是解决以上问题的FLask扩展。
看看它的__init__()函数：

  1. def __init__(self, app=None):
  2.         """If app argument provided then initialize celery using application config values.
  3.         If no app argument provided you should do initialization later with init_app method.
  4.         :param app: Flask application instance.
  5.         """
  6.         self.original_register_app = _state._register_app  # Backup Celery app registration function.
  7.         _state._register_app = lambda _: None  # Upon Celery app registration attempt, do nothing.
  8.         super(Celery, self).__init__()
  9.         if app is not None:
  10.             self.init_app(app)

将_state._register_app函数备份，再置为空。这样__init__()就不会创建Celery实例了。但如果指定了app，那么进入init_app，嗯，大多数Flask扩展都有这个函数，用来动态生成扩展实例。

  1. def init_app(self, app):
  2.         """Actual method to read celery settings from app configuration and initialize the celery instance.
  3.         :param app: Flask application instance.
  4.         """
  5.         _state._register_app = self.original_register_app  # Restore Celery app registration function.
  6.         if not hasattr(app, 'extensions'):
  7.             app.extensions = dict()
  8.         if 'celery' in app.extensions:
  9.             raise ValueError('Already registered extension CELERY.')
  10.         app.extensions['celery'] = _CeleryState(self, app)
  11.  
  12.         # Instantiate celery and read config.
  13.         super(Celery, self).__init__(app.import_name, broker=app.config['CELERY_BROKER_URL'])
  14.         ...

将_state._register_app函数还原，再执行Celery原本的__init__。这样就达到动态生成实例的目的了。接着往下看：

  1. task_base = self.Task
  2.  
  3. # Add Flask app context to celery instance.
  4. class ContextTask(task_base):
  5.     """Celery instance wrapped within the Flask app context."""
  6.     def __call__(self, *_args, **_kwargs):
  7.         with app.app_context():
  8.             return task_base.__call__(self, *_args, **_kwargs)
  9. setattr(ContextTask, 'abstract', True)
  10. setattr(self, 'Task', ContextTask)

这里重载了celery.Task类，通过with app.app_context():，在app.app_context()的上下文环境下执行Task。对于一个已生成的Flask实例，应用上下文不会随便改变。所以这就现实了在Celery中使用Flask的应用上下文。
下面是官方的示例代码：

  1. # extensions.py
  2. from flask_celery import Celery
  3. celery = Celery()
  4.  
  5. # application.py
  6. from flask import Flask
  7. from extensions import celery
  8.  
  9. def create_app():
  10.     app = Flask(__name__)
  11.     app.config['CELERY_IMPORTS'] = ('tasks.add_together', )
  12.     app.config['CELERY_BROKER_URL'] = 'redis://localhost'
  13.     app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost'
  14.     celery.init_app(app)
  15.     return app
  16.     
  17. # tasks.py
  18. from extensions import celery
  19.  
  20. @celery.task()
  21. def add_together(a, b):
  22.     return a + b
  23.  
  24. # manage.py
  25. from application import create_app
  26. app = create_app()
  27. app.run()

跟普通的Flask扩展一样了。

Celery中使用Flask上下文

在Flask的view函数中调用task.delay()时，这个task相当于一个离线的异步任务，它对Flask的应用上下文和请求上下文一无所知。但是这都可能是异步任务需要用到的。比如发送邮件要用到的render_template和url_for就分别要用到应用上下文和请求上下文。不在celery中引入它们的话，就是Running code outside of a request。
引入应用上下文的工作Flask-Celery-Helper已经帮我们做好了，在Flask的文档中也有相关介绍。实现方法和上面Flask-Celery-Helper的一样。然而，不管是Flask-Celery-Helper还是Flask文档，都没有提及如何在Celery中使用请求上下文。

要引入请求上下文，需要考虑这两个问题：

  1. 如何在Celery中产生请求上下文。Flask中有request_context和test_request_context可以产生请求上下文。区别是request_context需要WSGI环境变量environ，而test_request_context根据传入的参数生成请求上下文。我没有找到如何在Celery中获取到WSGI环境变量的方法，所以只能自己传入相关参数生成请求上下文了。

  2. 请求上下文是随HTTP请求产生的，要获取请求上下文，就必须在view函数中处理，view函数通过task.delay()发送Celery任务。所以需要重载task.delay()，以获取请求上下文。

具体的思路还是在init_app中重载celery.Task类，通过with app.test_request_context():，在app.test_request_context()的上下文环境下执行Task。
首先获取request，从中整理出test_request_context()需要的参数。根据test_request_context的函数注释，它需要的参数和werkzeug.test.EnvironBuilder类的参数一样。

  1. CONTEXT_ARG_NAME = '_flask_request_context'
  2. def _include_request_context(self, kwargs):
  3.     """Includes all the information about current Flask request context
  4.     as an additional argument to the task.
  5.     """
  6.     if not has_request_context():
  7.         return
  8.  
  9.     # keys correspond to arguments of :meth:`Flask.test_request_context`
  10.     context = {
  11.         'path': request.path,
  12.         'base_url': request.url_root,
  13.         'method': request.method,
  14.         'headers': dict(request.headers),
  15.         'data': request.form
  16.     }
  17.     if '?' in request.url:
  18.         context['query_string'] = request.url[(request.url.find('?') + 1):]
  19.  
  20.     kwargs[self.CONTEXT_ARG_NAME] = context

_include_request_context函数从request中提取path,base_url,method,headers,data,query_string。将他们传入test_request_context，生成伪造的请求上下文可以覆盖大多数的使用情况。
Celery通过apply_async,apply,retry调用异步任务（delay是apply_async的简化方法）。这里需要重载它们，让这些函数获取request：

  1. def apply_async(self, args=None, kwargs=None, **rest):
  2.     self._include_request_context(kwargs)
  3.     return super(ContextTask, self).apply_async(args, kwargs, **rest)
  4.  
  5. def apply(self, args=None, kwargs=None, **rest):
  6.     self._include_request_context(kwargs)
  7.     return super(ContextTask, self).apply(args, kwargs, **rest)
  8.  
  9. def retry(self, args=None, kwargs=None, **rest):
  10.     self._include_request_context(kwargs)
  11.     return super(ContextTask, self).retry(args, kwargs, **rest)

最后重载celery.Task的__call__方法：

  1. def __call__(self, *args, **kwargs):
  2.     """Execute task code with given arguments."""
  3.     call = lambda: super(ContextTask, self).__call__(*args, **kwargs)
  4.  
  5.     context = kwargs.pop(self.CONTEXT_ARG_NAME, None)
  6.     if context is None or has_request_context():
  7.         return call()
  8.  
  9.     with app.test_request_context(**context):
  10.         result = call()
  11.  
  12.         # process a fake "Response" so that
  13.         # ``@after_request`` hooks are executed
  14.         app.process_response(make_response(result or ''))
  15.  
  16.     return result

context是我们从request中获取的参数，将它传给test_request_context，伪造请求上下文，并在这个上下文环境中执行task。既然伪造了请求，那也得为这个假请求生成响应，万一你定义了after_request这个在响应后执行的钩子呢？通过process_response就可以激活after_request。
注意这里并没有传入应用上下文，因为Flask在创建请求上下文时，会判断应用上下文是否为空，为空就先创建应用上下文，再创建请求上下文。

完整代码在这里。
celery = CeleryWithContext()创建的Celery实例就可以给各种task使用了。
另外创建一个celery_worker.py文件，生成一个Flask实例，供Celery的worker使用。

  1. # celery_worker.py
  2.  
  3. #!/usr/bin/env python
  4. from app import create_app
  5. from app.extensions import celery
  6.  
  7. app = create_app()

启动worker：celery -A celery_worker.celery worker -l info
这下就可以使用Celery发邮件了。唉，还真是麻烦。

reference

http://xion.io/post/code/celery-include-flask-request-context.html

博客地址

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/凡人多烦事01/article/detail/414327?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号