Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
sqlalchemy invalidation test
from __future__ import print_function
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
#
# Use this file to build your own SSCCE
# SSCCE = Short, Self Contained, Correct (Compatible) Example
# see http://sscce.org/
#
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Standard imports
import pdb
import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import deferred, class_mapper
from sqlalchemy import Column, Integer
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import inspect
from sqlalchemy.orm.session import object_session
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# You probably don't need to overwrite this
Base = declarative_base()
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Define some models that inherit from Base
class Foo(Base):
__tablename__ = "foo"
id = Column(Integer, primary_key=True)
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# we'll use sqlite+memory to handle this
# if your issue is backend specific, then the engine will need to be different
sqlalchemy_url = "sqlite://"
# sqlalchemy_url = "postgresql://sa_test:sa_test@localhost/sqlalchemy_test"
engine = create_engine(sqlalchemy_url, echo=False)
Base.metadata.create_all(engine)
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# you shouldn't change these 2 line
sessionFactory = sessionmaker(bind=engine)
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# Start your example here:
# setup
for i in (1, 2):
try:
_s = sessionFactory()
f = Foo()
f.id = i
_s.add(f)
_s.commit()
except Exception as exc:
# duplicate errors likely in postgres
pass
def test_original():
"""
original testcase submitted in https://github.com/sqlalchemy/sqlalchemy/issues/5098
Under Sqlite:
1 conn= <sqlalchemy.engine.base.Connection object at 0x1022a0110>
2 s= <sqlalchemy.orm.session.Session object at 0x1022a0210>
3 r= <__main__.Foo object at 0x102704f90>
4 post invalidate()
Traceback (most recent call last):
sqlalchemy.exc.StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back
Under PostgreSQL
1 conn= <sqlalchemy.engine.base.Connection object at 0x1026e5050>
2 s= <sqlalchemy.orm.session.Session object at 0x102704250>
3 r= <__main__.Foo object at 0x102704f90>
4 post invalidate()
Traceback (most recent call last):
sqlalchemy.exc.StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back
"""
print("test_original")
conn = engine.connect()
print(1, "conn=", conn)
s = sessionFactory(bind=conn)
print(2, "s=", s)
r = s.query(Foo).filter_by(id=1).first()
print(3, "r=", r)
conn.invalidate()
print(4, "post invalidate()")
r = s.query(Foo).filter_by(id=1).first()
print(5, "r=", r)
def test_explore():
"""
"""
print("test_explore")
conn = engine.connect() # not necessary since Session will establish a connection
print(1, "conn=", conn)
s = sessionFactory(bind=conn)
print(2, "s=", s)
s_conn_1 = s.connection() # the session may have it's own connection
print(3, "s_conn_1=", s_conn_1)
r1 = s.query(Foo).filter_by(id=1).first()
print(4, "r1=", r1)
r1_info = inspect(r1)
print("r1_info.persistent", r1_info.persistent)
obj_conn_1 = object_session(r1).connection()
# make sure we have the same connection.
# an early testcase didn't supply `bind=conn` to the Session, which broke this test
assert (conn == s_conn_1) and (s_conn_1 == obj_conn_1)
if False:
print("object_session(r1).connection()", obj_conn_1)
print(
"object_session(r1).connection() == s.connection()", obj_conn_1 == s_conn_1
)
print("object_session(r1).connection() == engine.connect()", obj_conn_1 == conn)
# invalidate and close things
conn.invalidate()
obj_conn_1.invalidate()
print(5, "post invalidate()")
if False:
# under PostgreSQL, enabling this block will cause the query for `r2` to fail with the following error:
# sqlalchemy.exc.StatementError: (sqlalchemy.exc.ResourceClosedError) This Connection is closed
conn.close()
obj_conn_1.close()
print(6, "post close()")
print(7, "s=", s)
s_conn_2 = s.connection() # the session may have it's own connection
print(7, "s_conn_2=", s_conn_2)
try:
r2 = s.query(Foo).filter_by(id=2).first()
raise ValueError("this should have failed!")
except:
# expected!
raise
def test_concern():
"""
"""
print("test_concern")
s = sessionFactory() # just use the engine for auto-connections
print(2, "s=", s)
s_conn_1 = s.connection()
print(3, "s_conn_1=", s_conn_1)
r1 = s.query(Foo).filter_by(id=1).first()
print(4, "r1=", r1)
r1_object_connection = object_session(r1).connection()
print("5", "r1_object_connection=", r1_object_connection)
assert r1_object_connection == s_conn_1
# invalidate and close things
s_conn_1.invalidate()
if False:
# if this doesn't run...
# postgres/sqlite error on query below is:
# sqlalchemy.exc.StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back
# if this does run...
# postgres/sqlute error on query below is:
# sqlalchemy.exc.StatementError: (sqlalchemy.exc.ResourceClosedError) This Connection is closed
s_conn_1.close()
if False:
# running either of these is what we want to do so the test will run
s.close()
s.invalidate()
# now what happens if we keep going...
print(7, "s=", s)
s_conn_2 = s.connection() # the session may have it's own connection
print(8, "s_conn_2=", s_conn_2)
# todo: the setup data gets cleared away in sqlite. not sure why. example holds in postgresql
r2 = s.query(Foo).filter_by(id=2).first()
print(9, "r2=", r2)
r2_object_connection = object_session(r2).connection()
print("10", "r2_object_connection=", r2_object_connection)
assert r2_object_connection == s_conn_2
assert r2_object_connection != r1_object_connection
test_concern()
exit()
if True:
test_concern()
else:
test_original()
test_explore()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.