创建事务回滚操作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之外，还应该确保会话还没有在注册表中注册过。另外请注意，如果您有使用线程的代码，注册表将共享它们之间的连接，这会造成麻烦。