

@pytest.yield_fixture
def db_session(app_config):
    engine = app_config.registry['sqlalchemy.engine']
    conn = engine.connect()
    trans = conn.begin()
    session = Session(bind=conn)
    session.begin_nested()

    @event.listens_for(session, 'after_transaction_end')
    def restart_savepoint(session, transaction):
        if (transaction.nested and (not transaction._parent.nested)):
            session.begin_nested()
    try:
        (yield session)
    finally:
        session.close()
        Session.remove()
        trans.rollback()
        conn.close()
        engine.dispose()

dataset/ETHPy150Open pypa/warehouse/tests/conftest.py db_session/original