Python: SQLAlchemy – understanding sessions and associated queries. | behai-nguyen software development learnings and documentation

In this post, we look at some of the basics of sessionmaker, scoped_session and their associated query methods.


SQLAlchemy – understanding sessions and associated queries.

Please note that this post is not a tutorial. I sought to answer some of my own questions. And I’m merely writing down my answers.

The source database used in this post is the MySQL test data released by Oracle Corporation. Downloadable from https://github.com/datacharmer/test_db.

Let’s start with the Session class. SQLAlchemy official documentation Using the Session.

❶ We can use instances of sessionmaker to run full text queries:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://behai:pcb.2176310315865259@localhost/employees"

engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=False, future=True)

session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)

session = session_factory()

sql = "select * from employees where last_name like '%treh%' limit 0, 10"

dataset = session.execute(text(sql))
for r in dataset:
    print(r)

dataset.close()
session.close()

For future=True, please see SQLAlchemy 2.0 Future (Core)

❷ Next comes the scoped_session. Basically, it is the session that we should use in web applications: each scoped session is “local” to the context a web request. Please see Contextual/Thread-local Sessions.

For the purpose of this discussion, we will not be doing any web applications. Simple command line Python scripts would suffice for illustration purposes:

from threading import get_ident
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy import text

SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://behai:pcb.2176310315865259@localhost/employees"

engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=False, future=True)

session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)

database_sesssion = scoped_session(session_factory, scopefunc=get_ident)

session = database_sesssion(future=True)

sql = "select * from employees where last_name like '%treh%' limit 0, 10"

dataset = session.execute(text(sql))
for r in dataset:
    print(r)

dataset.close()
session.close()

This script does the same thing as the previous one, barring one addition, and one modification:

The addition is line 12:

database_sesssion = scoped_session(session_factory, scopefunc=get_ident)

For scopefunc=get_ident, please see sqlalchemy.orm.scoping.scoped_session, and this Stackoverflow post flask _app_ctx_stack.ident_func error due to ident_func deprecated in werkzeug 2.1 should make it clearer.

The modification is line 14:

session = database_sesssion(future=True)

In this second script, instead of getting a session directly from sessionmaker, we get one indirectly from scoped_session. We are running the same query as the previous script, so the final output is also the same.

From here on, scripts all have lines 1-12 ( one to twelve ) identical, I will only list the relevant new codes for the current discussions.

❸ According to Contextual/Thread-local Sessions, after:

session = database_sesssion(future=True)

If we make repeated calls to database_sesssion(), we get back the same session:

...
session = database_sesssion(future=True)
session1 = database_sesssion()

print(f"1. database_sesssion: {id(database_sesssion)}")
print(f"1. session: {id(session)}")
print(f"1. session1: {id(session1)}")
print(f"1. session is session1: {session is session1}")

database_sesssion: 1724058633408
session: 1724061992896
session1: 1724061992896
session is session1: True

We can call database_sesssion(future=True) only once, subsequent calls must be made with no parameters, otherwise it will result in the following exception:

(venv) F:\my_project>venv\Scripts\python.exe src\my_project\my_python_script.py
Traceback (most recent call last):
  File "F:\my_project\src\my_project\my_python_script.py", line 15, in <module>
    session1 = database_sesssion(future=True)
  File "F:\my_project\venv\lib\site-packages\sqlalchemy\orm\scoping.py", line 39, in __call__
    raise sa_exc.InvalidRequestError(
sqlalchemy.exc.InvalidRequestError: Scoped session is already present; no new arguments may be specified.

❹ According to method sqlalchemy.orm.scoping.scoped_session.remove():

Dispose of the current Session, if present.

This will first call Session.close() method on the current Session, which releases any existing transactional/connection resources still being held; transactions specifically are rolled back. The Session is then discarded. Upon next usage within the same scope, the scoped_session will produce a new Session object.

Let’s see what that means, the following script produces the same results as the first script:

...
session = database_sesssion(future=True)
session.close()

database_sesssion.remove()

session = database_sesssion(future=True)

sql = "select * from employees where last_name like '%treh%' limit 0, 10"

dataset = session.execute(text(sql))
for r in dataset:
    print(r)

dataset.close()
session.close()

After calling database_sesssion.remove(), a subsequent call with parameter session = database_sesssion(future=True) actually works, it does not raise an exception.

This makes sense in the context of the above statement: the internal registry is now empty, there is no active scoped session present, so we can create a new one with whatever configurations we see fit.

This would suggest that the database_sesssion itself is still the same object after calling database_sesssion.remove(). Let’s test it out with the following script:

...
session = database_sesssion(future=True)
session1 = database_sesssion()

print(f"1. database_sesssion: {id(database_sesssion)}")
print(f"1. session: {id(session)}")
print(f"1. session1: {id(session1)}")
print(f"1. session is session1: {session is session1}")

session1.close()
session.close()

database_sesssion.remove()

session = database_sesssion(future=True)
session1 = database_sesssion()

print(f"2. database_sesssion: {id(database_sesssion)}")
print(f"2. session: {id(session)}")
print(f"2. session1: {id(session1)}")

We can see that it is, in the output:

database_sesssion: 2102627796160
session: 2102631155648
session1: 2102631155648
session is session1: True
database_sesssion: 2102627796160
session: 2102631155792
session1: 2102631155792

❺ We will introduce a model, which basically is a SQLAlchemy class representation of a database table. As per documentation, we will descend our model from function sqlalchemy.orm.declarative_base(…), then we can use method sqlalchemy.orm.scoping.scoped_session.query_property(query_cls=None) with this model. We will do a model for the employees table.

❺⓵ Despite the official document example, I wanted to try using database_sesssion.query_property() directly, this is my first attempt ( I’m listing a complete new script ):

from threading import get_ident
from sqlalchemy import (
    create_engine,
    Column,
    Integer,
    Date,
    String,
)
from sqlalchemy.orm import sessionmaker, scoped_session
"""
Any one of these import of declarative_base will work.
"""
# from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import declarative_base

SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://behai:pcb.2176310315865259@localhost/employees"

engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=False, future=True)

session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)

database_sesssion = scoped_session(session_factory, scopefunc=get_ident)

class Employees(declarative_base()):
    __tablename__ = 'employees'

    emp_no = Column(Integer, primary_key=True)
    birth_date = Column(Date, nullable=False)
    first_name = Column(String(14), nullable=False)
    last_name = Column(String(16), nullable=False)
    gender = Column(String(1), nullable=False )
    hire_date = Column(Date, nullable=False )

query = database_sesssion.query_property()
result = query.filter(Employees.emp_no==16621).first()

I did expect it to work, but it did not:

(venv) F:\my_project>venv\Scripts\python.exe src\my_project\my_python_script.py
Traceback (most recent call last):
  File "F:\my_project\src\my_project\my_python_script.py", line 35, in <module>
    result = query.filter(Employees.emp_no==16621).first()
AttributeError: 'query' object has no attribute 'filter'

From here on, scripts will have lines 1-22 ( one to twenty two ) identical to ❺⓵, I will only list the relevant additions and changes for the current discussions.

❺⓶ The example snippet listed under method sqlalchemy.orm.scoping.scoped_session.query_property(query_cls=None) is:

<pre>
Session = scoped_session(sessionmaker())

class MyClass(object):
    query = Session.query_property()

# after mappers are defined
result = MyClass.query.filter(MyClass.name=='foo').all()

Accordingly, I modify ❺⓵ as follows:

...
class BaseModel(object):
    query = database_sesssion.query_property()

Base = declarative_base(cls=BaseModel)

class Employees(Base):
    __tablename__ = 'employees'

    ...

result = Employees.query.filter(Employees.emp_no==16621).first()
print(result.__dict__)

And it does work as expected:

(venv) F:\my_project>venv\Scripts\python.exe src\my_project\my_python_script.py
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x0000026A680079A0>, 'last_name': 'Strehl', 'emp_no': 16621, 'hire_date': datetime.date(1992, 6, 11), 'first_name': 'Parviz', 'gender': 'M', 'birth_date': datetime.date(1962, 5, 30)}

I don’t know why it has to be like this. But the official document states that it must be, and my experimentation confirms it. I’m not going to dig into the SQLAlchemy codes to see why! I have a feeling that it’s going to be futile, too 😂…

We’ve seen previously that scoped_session(…)([…]) and sessionmaker(…)() both result in a class sqlalchemy.orm.Session(…), and this class has a method sqlalchemy.orm.Session.query(*entities, **kwargs). This method can be used to do what scoped_session(…).query_property() does, barring some differences. The first is that the model can descend directly from declarative_base(). Secondly, the syntax is slightly different.

❺⓷The modified script which uses scoped_session(…)([…])’s query() method:

...
class Employees(declarative_base()):
    __tablename__ = 'employees'

    ...

session = database_sesssion(future=True)
result = session.query(Employees).filter(Employees.emp_no==10545).first()
print( result.__dict__ )

❺⓸ And similarly for sessionmaker(…)():

...
class Employees(declarative_base()):
    __tablename__ = 'employees'

    ...

session = session_factory()
result = session.query(Employees).filter(Employees.emp_no==11000).first()
print( result.__dict__ )

Let’s reiterate that we use scoped_session with web applications – Contextual/Thread-local Sessions. We include sessionmaker for the shake of comparison and completeness. Back to scoped_session’s query() method and query property:

● ❺⓶: result = Employees.query.filter(Employees.emp_no==16621).first()

● ❺⓷: result = session.query(Employees).filter(Employees.emp_no==10545).first()

I choose to use ❺⓶ approach, despite having to have an extra base class. I.e.:

class BaseModel(object):
    query = database_sesssion.query_property()

This BaseModel can be the parent class for all of the database tables, so in the long run, it sort of pays for itself.

✿✿✿

As I have stated up front, this post is not a tutorial. I find SQLAlchemy to be a tough row to hoe… I set out to answer my own questions. I hope the content of this post will be useful for others who are learning SQLAlchemy. I do hope I have not made any mistakes in this post. Thank you for reading and stay safe as always.