我正在尝试建立一个系统,将数据库操作优雅地推迟到单独的线程,以避免在 Twisted 回调期间发生阻塞。
到目前为止,这是我的方法:
from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from twisted.internet.threads import deferToThread
_engine = create_engine(initialization_string)
Session = scoped_session(sessionmaker(bind=_engine))
@contextmanager
def transaction_context():
session = Session()
try:
yield session
session.commit()
except:
# No need to do session.rollback(). session.remove will do it.
raise
finally:
session.remove()
def threaded(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
return deferToThread(fn, *args, **kwargs)
return wrapper
这应该允许我用以下内容包装一个函数threaded
装饰器,然后使用transaction_context
所述函数体内的上下文管理器。下面是一个例子:
from __future__ import print_function
from my_lib.orm import User, transaction_context, threaded
from twisted.internet import reactor
@threaded
def get_n_users(n):
with transaction_context() as session:
return session.query(User).limit(n).all()
if __name__ == '__main__':
get_n_users(n).addBoth(len)
reactor.run()
但是,当我运行上述脚本时,出现包含以下回溯的失败:
Unhandled error in Deferred:
Unhandled Error
Traceback (most recent call last):
File "/usr/lib/python2.7/threading.py", line 781, in __bootstrap
self.__bootstrap_inner()
File "/usr/lib/python2.7/threading.py", line 808, in __bootstrap_inner
self.run()
File "/usr/lib/python2.7/threading.py", line 761, in run
self.__target(*self.__args, **self.__kwargs)
--- <exception caught here> ---
File "/usr/local/lib/python2.7/dist-packages/twisted/python/threadpool.py", line 191, in _worker
result = context.call(ctx, function, *args, **kwargs)
File "/usr/local/lib/python2.7/dist-packages/twisted/python/context.py", line 118, in callWithContext
return self.currentContext().callWithContext(ctx, func, *args, **kw)
File "/usr/local/lib/python2.7/dist-packages/twisted/python/context.py", line 81, in callWithContext
return func(*args,**kw)
File "testaccess.py", line 9, in get_n_users
return session.query(User).limit(n).all()
File "/usr/lib/python2.7/contextlib.py", line 24, in __exit__
self.gen.next()
File "/home/louis/Documents/Python/knacki/knacki/db.py", line 36, in transaction_context
session.remove()
exceptions.AttributeError: 'Session' object has no attribute 'remove'
我根本没想到会这样。我缺少什么?我没有实例化我的scoped_session
适当地?
Edit: Here https://stackoverflow.com/questions/21076105/is-this-an-acceptable-way-to-make-threaded-sqlalchemy-queries-from-twisted是有关将此设置与 Twisted 集成的相关问题。它可能有助于澄清我想要实现的目标。