2

我正在尝试设置 Celery 任务。我们的主要应用是带有 SQLAlchemy 的 Pyramid。

所以我有一个任务定义为：

from celery.contrib.methods import task
from apipython.celerytasks import celery

class Email():
    def __init__(self, from_name, from_email, to_name, to_email, subject, html_body,
                 sendgrid_category=None):
        self.from_name = from_name
        self.from_email = from_email
        self.to_name = to_name
        self.to_email = to_email
        self.subject = subject
        self.body = None
        self.html_body = html_body
        self.sendgrid_category = sendgrid_category

class EmailService():
    @task()
    def task__send_smtp(self, email, from_user_id=None, to_user_id=None):
        # send the email, not shown here

        # EmailLog is a SQLAlchemy model
        email_log = EmailLog(
                        email.subject,
                        email.html_body,
                        from_user_id=from_user_id,
                        to_user_id=to_user_id,
                        action_type=email.sendgrid_category)
        DBSession.add(email_log)

        transaction.commit()

还有 celerytasks.py 我有：

from celery import Celery

celery = Celery('apipython.celery',
                broker='sqla+mysql+mysqldb://root:notarealpassword@127.0.0.1/gs?charset=utf8',
                backend=None,
                include=['apipython.services.NotificationService'])

if __name__ == '__main__':
    celery.start()

它有效 - 任务被序列化并被拾取。

但是，当我尝试在任务中使用 SQLAlchemy / DBSession 时，出现错误：

UnboundExecutionError: Could not locate a bind configured on mapper Mapper|EmailLog|emaillogs or this Session

我了解工作任务在单独的进程上运行，需要设置其设置、会话、引擎等。所以我有这个：

@worker_init.connect
def bootstrap_pyramid(signal, sender):
    import os
    from pyramid.paster import bootstrap
    sender.app.settings = bootstrap('development.ini')['registry'].settings

    customize_settings(sender.app.settings)

    engine = sqlalchemy.create_engine('mysql+mysqldb://root:notarealpassword@127.0.0.1/gs?charset=utf8')
    DBSession.configure(bind=engine)
    Base.metadata.bind = engine

但是我仍然遇到同样的错误。

DBSession 和 Base 在 models.py 中定义为

DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
Base = declarative_base()

为了使模型绑定起作用，我缺少什么步骤？

第二个问题，这个用于创建会话/绑定的代码可以在 celery 的 init 和 worker init 中工作吗？

（顺便说一句，我确实尝试过 pyramid_celery 但更喜欢制作普通的芹菜）

谢谢，

4

1 回答 1

1

我的同事尝试了完全相同的代码，并且成功了。奇怪的

于 2013-11-09T05:16:27.880 回答