SQLAlchemy not identifying Python pool threads as seperate process

user_3579955 0
Viewed 146

I am trying to convert my single threaded application to multi threaded application which uses database using SQLAlchemy. And I found that SQLAlchemy session is not thread safe. So we need to use scoped_session factory for thread safe db access.

Below is my input dataset

input_list = [data1, data2, data3, data4, data5]

Single thread application

from sqlalchemy.orm import sessionmaker, scoped_session
Session = sessionmaker(bind=engine_url)

for data in input_list:
    def myfunction(data):
        db_session = Session()
        print(db_session)
        # use db_session to query/store the data

When I try to convert it to multithreaded application

from sqlalchemy.orm import sessionmaker, scoped_session
Session = scoped_session(sessionmaker(bind=engine_url))

def myfunction(data):
    db_session = Session()
    print(db_session)
    # use db_session to query/store the data

def myfunction_parallel():
    with ThreadPool(4) as pool:
        output = pool.map(myfunction, input_list)

In multithread variant, I am getting db_session as same object, but my expectation is that there should be a new session object created for each thread and the session should be different?

sqlalchemy multithreading python database
1 Answers

The scoped session registry registers session for each thread that requests one. This enables code to call db_session = Session() and get the expected session for the thread.

However it's application's responsibility to inform the session registry when a session is no longer required. The application does this by calling Session.remove(), as documented here:

The scoped_session.remove() method first calls Session.close() on the current Session, which has the effect of releasing any connection/transactional resources owned by the Session first, then discarding the Session itself. “Releasing” here means that connections are returned to their connection pool and any transactional state is rolled back, ultimately using the rollback() method of the underlying DBAPI connection.

At this point, the scoped_session object is “empty”, and will create a new Session when called again.

This code should work as expected:

def myfunction(data):
    db_session = Session()
    print(db_session)
    # use db_session to query/store the data
    Session.remove()
user_53209060
Related