python - Celery 任务中的 Pyramid/SQLAlchemy 模型绑定(bind)

Question

我正在尝试设置 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

我知道工作任务在一个单独的进程上运行，需要设置它的设置、 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()

要使模型绑定(bind)正常工作，我缺少哪一步？

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

(顺便说一句，我确实尝试过 pyramid_celery 但更喜欢制作普通 celery )

谢谢，

Answer 1

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