Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
sqlalchemy test cases for bug in listen() on Engine events fired on the Connection
import sqlalchemy
from sqlalchemy import create_engine, Column, Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
def setup():
# set up a simple in memory sqlite db with two table
engine = create_engine('sqlite:///:memory:')
Base = declarative_base(bind=engine)
class A(Base):
__tablename__ = "a"
id = Column("id", Integer, primary_key = True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
return engine, A, Session
def test_session_before_listen():
engine, A, Session = setup()
got_event = []
def before_execute_cb(*args):
got_event.append(True)
sess = Session()
sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
sess.query(A).all()
# OK
assert len(got_event) == 1
def test_query_before_listen():
engine, A, Session = setup()
got_event = []
def before_execute_cb(*args):
got_event.append(True)
sess = Session()
sess.query(A).all()
sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
sess2 = Session()
sess2.query(A).all()
# OK
assert len(got_event) == 1
def test_query_before_listen_same_session():
engine, A, Session = setup()
got_event = []
def before_execute_cb(*args):
got_event.append(True)
sess = Session()
sess.query(A).all()
sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
sess.query(A).all() # <--- USE THE SAME SESSION AS BEFORE
# FAIL
assert len(got_event) == 1
def test_query_with_pre_registered_dummy_listener():
engine, A, Session = setup()
got_event = []
def before_execute_cb(*args):
got_event.append(True)
def foo(*args):
pass
sqlalchemy.event.listen(engine, "before_execute", foo) # <--- Register a dummy listener to force sqlalchemy to join the Connection's dispatch descriptor with the Engine's
sess = Session()
sess.query(A).all()
sqlalchemy.event.listen(engine, "before_execute", before_execute_cb)
sess.query(A).all() # <--- USE THE SAME SESSION AS BEFORE
# OK
assert len(got_event) == 1
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.