創建事務回滾操作DB圍繞這一`Manager`結構

Question

我創建圍繞SQLAlchemy的數據庫引擎/連接/會話中的「經理」對象：

Base = declarative_base() 

class Manager(object): 
    def __init__(self, connection: str = 'sqlite://'): 
     self.engine = create_engine(connection, echo=True) 
     Base.metadata.create_all(self.engine) 
     self.sessionmaker = sessionmaker(bind=self.engine) 
     self.session = scoped_session(self.sessionmaker) 

    def do_db_stuff(self): 
     self.session.query(Whatever).all() 

    def ensure_thing(self): 
     thing = Thing() 
     self.session.add(thing) 
     self.session.commit() 

我想創建兩個py.test fixtures：一個實例化管理器，一個在可能調用commit的測試中打包和回滾事務。 This is the pattern我試圖跟隨，但沒有成功：即使通過上述棒調用transaction.rollback()後周圍環繞Manager創建

@pytest.fixture(scope='session') 
def manager(): 
    m = Manager() 
    return m 


@pytest.fixture(scope='function') 
def manager_session(manager): 
    connection = manager.session.connection() 
    transaction = connection.begin() 

    yield manager 

    manager.session.close() 
    transaction.rollback() 
    connection.close() 

不幸的是，對象。

在這樣的現有會話周圍包裝事務的正確方法是什麼？

編輯：

另外，不同的嘗試：

@pytest.fixture(scope='function') 
def manager_session(manager): 
    connection = manager.engine.connect() 
    transaction = connection.begin() 
    manager.sessionmaker.configure(bind=connection) 

    yield manager 

    manager.session.close() 
    transaction.rollback() 

編輯2：

，似乎工作，在Ilja Everilä's answer below提到需要提醒的是線程代碼會造成麻煩第三次嘗試。

@pytest.fixture(scope='session') 
def manager(): 
    return Manager() 


@pytest.fixture(scope='function') 
def manager_transaction(manager): 
    connection = manager.engine.connect() 
    transaction = connection.begin() 
    manager.session_maker.configure(bind=connection) 

    yield manager 

    manager.session_maker.configure(bind=manager.engine) 
    manager.session.remove() 
    transaction.rollback() 
    connection.close() 

Answer 1

第一次嘗試失敗，因爲會話實際上是在控制連接及其事務。您可以通過查看生成的日誌記錄來驗證。當您撥打manager.session.connection()並顯式調用begin()是一個無操作返回正在進行的事務對象時，會話將開始一個隱式事務。因此，當您在管理器方法中提交時，您會提交真實的，並且當您回滾時，過時的事務對象不會執行任何操作。

第二次嘗試對我有用如同一樣，如果使用內存中的SQLite數據庫，但如果您的實際代碼與您所呈現的略有不同，則不起作用。您將創建的連接設置爲self.sessionmaker上的綁定，而不是已創建的scoped session registryself.session中的會話，因此如果在配置製造商之前以任何方式觸摸了會話註冊表，則實際上已使用引擎創建會話作爲綁定在當前線程中：

In [7]: m = Manager() 

In [8]: m.session.bind 
Out[8]: Engine(sqlite://) 

In [9]: connection = m.engine.connect() 

In [10]: transaction = connection.begin() 
2017-08-28 14:24:02,584 INFO sqlalchemy.engine.base.Engine BEGIN (implicit) 

In [11]: m.sessionmaker.configure(bind=connection) 

In [12]: m.session.bind 
Out[12]: Engine(sqlite://) 

所以除了配置sessionmaker之外，還應該確保會話還沒有在註冊表中註冊過。另外請注意，如果您有使用線程的代碼，註冊表將共享它們之間的連接，這會造成麻煩。