带有create_app，SQLAlchemy和Celery的Flask

2 回答

• 14

  我关注了current_app的建议 .

  您的celery对象需要访问应用程序上下文 . 我在网上找到了一些关于使用工厂函数创建Celery对象的信息 . 在没有消息代理的情况下测试以下示例 .

  #factory.py
  from celery import Celery
  from config import config
  
  def create_celery_app(app=None):
      app = app or create_app(config)
      celery = Celery(__name__, broker=app.config['CELERY_BROKER_URL'])
      celery.conf.update(app.config)
      TaskBase = celery.Task
  
      class ContextTask(TaskBase):
          abstract = True
  
          def __call__(self, *args, **kwargs):
              with app.app_context():
                  return TaskBase.__call__(self, *args, **kwargs)
  
      celery.Task = ContextTask
      return celery
  

  并在tasks.py中：

  #tasks.py
  from factory import create_celery_app
  from celery.signals import task_prerun
  from flask import g
  
  celery = create_celery_app()
  
  @task_prerun.connect
  def celery_prerun(*args, **kwargs):
      #print g
      with celery.app.app_context():
      #   # use g.db
         print g
  
  @celery.task()
  def do_some_stuff():
      with celery.app.app_context():
          # use g.db
          g.user = "test"
          print g.user
  

  一些链接：

  Flask pattern for creating a Celery instance with factory function

  Application using both application factory and celery

  Source for said application's factory.py

  Source for application tasks.py

  回复于 2024-04-26T06:18:46+08:00
• 2

  这是一个解决方案，它适用于烧瓶应用工厂模式，并且还可以使用上下文创建芹菜任务，而无需在任务中明确使用 app.app_context() . 在我的应用程序中，获取该app对象同时避免循环导入是非常棘手的，但这解决了它 . 在撰写本文时，这也适用于最新的芹菜版本4.2 .

  结构体：

  repo_name/
      manage.py
      base/
      base/__init__.py
      base/app.py
      base/runcelery.py
      base/celeryconfig.py
      base/utility/celery_util.py
      base/tasks/workers.py
  

  所以 base 是本例中的主要应用程序包 . 在 base/__init__.py 中我们创建芹菜实例如下：

  from celery import Celery
  celery = Celery('base', config_source='base.celeryconfig')
  

  base/app.py 文件包含烧瓶应用工厂 create_app 并注意它包含的 init_celery(app, celery) ：

  from base import celery
  from base.utility.celery_util import init_celery
  
  def create_app(config_obj):
      """An application factory, as explained here:
      http://flask.pocoo.org/docs/patterns/appfactories/.
      :param config_object: The configuration object to use.
      """
      app = Flask('base')
      app.config.from_object(config_obj)
      init_celery(app, celery=celery)
      register_extensions(app)
      register_blueprints(app)
      register_errorhandlers(app)
      register_app_context_processors(app)
      return app
  

  继续 base/runcelery.py 内容：

  from flask.helpers import get_debug_flag
  from base.settings import DevConfig, ProdConfig
  from base import celery
  from base.app import create_app
  from base.utility.celery_util import init_celery
  CONFIG = DevConfig if get_debug_flag() else ProdConfig
  app = create_app(CONFIG)
  init_celery(app, celery)
  

  接下来， base/celeryconfig.py 文件（作为示例）：

  # -*- coding: utf-8 -*-
  """
  Configure Celery. See the configuration guide at ->
  http://docs.celeryproject.org/en/master/userguide/configuration.html#configuration
  """
  
  ## Broker settings.
  broker_url = 'pyamqp://guest:guest@localhost:5672//'
  broker_heartbeat=0
  
  # List of modules to import when the Celery worker starts.
  imports = ('base.tasks.workers',)
  
  ## Using the database to store task state and results.
  result_backend = 'rpc'
  #result_persistent = False
  
  accept_content = ['json', 'application/text']
  
  result_serializer = 'json'
  timezone = "UTC"
  
  # define periodic tasks / cron here
  # beat_schedule = {
  #    'add-every-10-seconds': {
  #        'task': 'workers.add_together',
  #        'schedule': 10.0,
  #        'args': (16, 16)
  #    },
  # }
  

  现在在 base/utility/celery_util.py 文件中定义init_celery：

  # -*- coding: utf-8 -*-
  
  def init_celery(app, celery):
      """Add flask app context to celery.Task"""
      TaskBase = celery.Task
      class ContextTask(TaskBase):
          abstract = True
          def __call__(self, *args, **kwargs):
              with app.app_context():
                  return TaskBase.__call__(self, *args, **kwargs)
      celery.Task = ContextTask
  

  对于 base/tasks/workers.py 中的 Worker ：

  from base import celery as celery_app
  from flask_security.utils import config_value, send_mail
  from base.bp.users.models.user_models import User
  
  @celery_app.task
  def send_welcome_email(email, user_id, confirmation_link):
      """Background task to send a welcome email with flask-security's mail.
      You don't need to use with app.app_context() as Task has app context.
      """
      user = User.query.filter_by(id=user_id).first()
      print(f'sending user {user} a welcome email')
      send_mail(config_value('EMAIL_SUBJECT_REGISTER'),
                email,
                'welcome', user=user,
                confirmation_link=confirmation_link) 
  
  @celery_app.task
  def do_some_stuff():
      print(g)
  

  然后，您需要从 repo_name 文件夹内的两个不同的cmd提示中启动芹菜殴打和芹菜工作者 .

  在一个cmd提示符下执行 celery -A base.runcelery:celery beat ，另一个 celery -A base.runcelery:celery worker .

  然后，执行需要烧瓶上下文的任务 . 应该管用 .

  回复于 2024-04-26T06:18:46+08:00