在中使用sqlalchemy作用域的\u会话螺纹

Question

我在使用sqlalchemy和线程时遇到问题。在

import queue
import threading

import sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, Sequence
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.scoping import scoped_session

engine = create_engine('sqlite:///:memory:', echo=False)
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)

    def __repr__(self):
        return "<User(name='%s', fullname='%s', password='%s')>" % (
        self.name, self.fullname, self.password)
Base.metadata.create_all(engine)

sessionfactory = sessionmaker(bind=engine)

# called by each thread
def write_name(q, name, sessionfactory):
    session = scoped_session(sessionfactory)
    ed_user = User(name=name, fullname='Power', password='edspassword')
    session.add(ed_user)
    session.commit()
    q.put(name)

names = ["Max", "Austin"]

q = queue.Queue()

for u in names:
    t = threading.Thread(target=write_name, args = (q, u, sessionfactory))
    t.daemon = True
    t.start()

s = q.get()

这将导致：

^{pr2}$

但是在主线程中添加和读取数据可以很好地工作。此外，我假设我需要在多进程上使用线程，因为作用域的\u会话使用线程本地存储。在

Answer 1

主要问题是你不能有多个连接到只存在于内存中的SQLite数据库，因为每个连接都会创建一个新的空数据库。请参阅SQLAlchemy docs。简而言之，您需要像这样创建引擎，以确保只有一个实例可以跨线程共享。在

from sqlalchemy.pool import StaticPool
engine = create_engine('sqlite://:memory:',
    connect_args={'check_same_thread': False},
    poolclass=StaticPool, echo=True)

一旦你这样做了，你就不需要scoped_session，因为scoped_session的目的是为每个线程创建一个连接，而你在这里不能这样做。在

另外，请注意，如果您希望一个scoped_session实例正常工作（使用非SQLite引擎）。你应该把它当作全局变量来处理，这样它就可以处理线程的局部事务了。在