Using sqlalchemy scoped_session in theading.Thread

Question

I'm having problems using sqlalchemy together with threading.

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()

This results in:

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: users [SQL: 'INSERT INTO users (name, fullname, password) VALUES (?, ?, ?)'] [parameters: ('Max', 'Power', 'edspassword')]

but it works fine to add and read data in the main thread. Furthermore I assume I need to use threading over multiprocess because scoped_session uses thread local storage.

Answer 1

The main problem is that you can't have multiple connections to a SQLite database that only exists in memory, because each connection will create a new empty database. See the SQLAlchemy docs on this. In short, you need to create the engine like this to make sure that is only one instance that can be shared across threads.

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

Once you do that, you don't need scoped_session, because the point of scoped_session is to create one connection per each thread and you specifically can't do that here.

Also, note that you should have only one scoped_session instance if you want it to work correctly (with a non-SQLite engine). You should treat it as global variable and then it will be able to handle the thread-local stuff.