Celery 任务中的 Pyramid / SQLAlchemy 模型绑定答案

【问题标题】：Pyramid / SQLAlchemy model binding in Celery tasksCelery 任务中的 Pyramid / SQLAlchemy 模型绑定
【发布时间】：2013-11-09 13:16:27
【问题描述】：

我正在尝试设置 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 但更喜欢制作普通的芹菜）

谢谢，

【问题讨论】：

您没有在此处显示所有代码，但是，您传递给任务的 email 是什么？那是一个 SQLAlchemy 对象吗？您不能在会话之间共享 SQLAlchemy 对象。
@AntoineLeclair Email 是一个普通的 python 对象，而不是 SQLAlchemy 对象。在任务中，我正在创建一个新的 EmailLog 实例，它是一个 SQLAlchemy 模型，并尝试使用 DBSession.add 和 transaction.commit。
重要的是你如何定义你的会话？你用scoped_session了吗？
嗨@javex，是的，我在models.py中有这两行 DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension())) Base = declarative_base()
您是否确保引导部分实际执行正确？尝试在引导部分使用会话。配置后它也会在那里抛出吗？

标签： python sqlalchemy celery pyramid

【解决方案1】：

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

【讨论】：