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

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

谢谢，

请您参考如下方法:

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