我尝试按照这个基于芹菜的后台任务为一个简单的应用程序创建芹菜设置。
在我的task.py
from celery import Celery
def make_celery(app):
celery = Celery(app.import_name, backend=app.config['CELERY_RESULT_BACKEND'],
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
该方法适用于主烧瓶应用的app.py
。
from flask import Flask
flask_app = Flask(__name__)
flask_app.config.update(
CELERY_BROKER_URL='redis://localhost:6379',
CELERY_RESULT_BACKEND='redis://localhost:6379'
)
celery = make_celery(flask_app)
@celery.task()
def add_together(a, b):
return a + b
我的用例是我想创建另一个模块
helpers.py
我 可以定义异步类的集合。分离 基于芹菜的方法并使其模块化。
我所做的是将task.py
模块调用到其他模块helpers.py
,以便创建一个类AsyncMail
来处理电子邮件操作后台工作。
from task import make_celery
class AsyncMail(object):
def __init__(self, app):
"""
:param app: An instance of a flask application.
"""
self.celery = make_celery(app)
def send(self, msg):
print(msg)
现在,我如何访问self.celery
属性以成为类的任何方法的装饰器?
@celery.task()
def send(self, msg):
print(msg)
如果不可能,还有什么其他替代步骤可以解决这个问题?
你不能做你想做的事情。在定义类时,没有self
,更不用说self.celery
调用了,所以你不能使用@self.celery
。即使你有某种时间机器,也可以创建 38 个不同的AsyncMail
实例,你想要哪一个self.celery
?
在讨论如何做你想做的事之前,你确定要做吗?你真的希望每个AsyncMail
对象都有自己的单独的芹菜吗?通常每个应用程序只有一个,这就是为什么通常不会出现这种情况的原因。
如果你真的想,你可以在有一个对象来装饰每个实例之后给它们装饰方法。但它会很丑陋。
def __init__(self, app):
self.celery = make_celery(app)
# We need to get the function off the class, not the bound method off self
send = type(self).send
# Then we decorate it manually—this is all @self.celery.task does
send = self.celery.task(send)
# Then we manually bind it as a method
send = send.__get__(self)
# And now we can store it as an instance attribute, shadowing the class's
self.send = send
或者,如果您希望将它们放在一行中:
self.send = self.celery.task(type(self).send).__get__(self)
对于 Python 2,"函数关闭类"实际上是一个未绑定的方法,IIRC 你必须调用__get__(self, type(self))
才能在最后将其转换为绑定方法,否则它应该都是一样的。