Ditch scoped session stuff, use orm-independent transaction instead

Author: bharling

example_app/db.py

@@ -5,10 +5,13 @@
 from sqlalchemy.orm import scoped_session, sessionmaker
 import os
 
-Base = declarative_base()
-Session = scoped_session(sessionmaker(expire_on_commit=False))
 engine = create_engine(os.environ['DATABASE_URL'])
-Session.configure(bind=engine)
+
+Base = declarative_base(bind=engine)
+
+Session = sessionmaker(expire_on_commit=False, bind=engine)
+
+session = Session()
 
 
 class User(Base):

pytest_sqlalchemy.py

@@ -158,14 +158,12 @@ def sqlalchemy_session():
 
 
 @pytest.fixture(scope='session')
-def connection(test_engine, sqlalchemy_base, sqlalchemy_session):
+def connection(request, test_engine, sqlalchemy_base, sqlalchemy_session):
 
     sqlalchemy_base.metadata.bind = test_engine
     sqlalchemy_base.metadata.create_all(test_engine)
 
     with test_engine.connect() as conn:
-        sqlalchemy_session.registry.clear()
-        sqlalchemy_session.configure(bind=conn)
         yield conn
 
     sqlalchemy_base.metadata.drop_all()
@@ -174,11 +172,11 @@ def connection(test_engine, sqlalchemy_base, sqlalchemy_session):
 @pytest.fixture
 def db_session(request, connection, sqlalchemy_session, monkeypatch):
 
-    monkeypatch.setattr(sqlalchemy_session, 'commit', sqlalchemy_session.flush)
-    monkeypatch.setattr(sqlalchemy_session, 'remove', lambda: None)
+    transaction = connection.begin()
+    sqlalchemy_session.bind = connection
 
-    @request.addfinalizer
-    def finalize():
-        sqlalchemy_session.rollback()
+    yield sqlalchemy_session
+
+    transaction.rollback()
+    sqlalchemy_session.close()
 
-    return sqlalchemy_session()

tests/conftest.py

@@ -1,12 +1,12 @@
 import pytest
 pytest_plugins = 'pytester'
 
-from example_app.db import Base, Session
+from example_app.db import Base, session
 
 @pytest.fixture(scope='session')
 def sqlalchemy_base():
     return Base
 
 @pytest.fixture(scope='session')
 def sqlalchemy_session():
-    return Session
+    return session

tests/test_db_fixture.py

@@ -6,5 +6,16 @@ def test_create_user(db_session):
     db_session.flush()
     assert db_session.query(User).count() == 1
 
+def test_update_object(db_session):
+    user = User(name='The User')
+    db_session.add(user)
+    db_session.flush()
+    assert db_session.query(User).count() == 1
+
+    user.name = 'Bongo'
+    db_session.commit()
+
+    db_session.query(User).filter(User.name=='Bongo').one()
+
 def test_tests_run_in_transactions(db_session):
     assert db_session.query(User).count() == 0