问题描述
我正在学习“Cosmic Python”一书,第 6 章解释了如何使用工作单元模式来改变与数据库/存储库的交互。
可在此处访问本书的第 6 章: https://www.cosmicpython.com/book/chapter_06_uow.html
作者提供的代码如下:
from __future__ import annotations
import abc
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.session import Session
from allocation import config
from allocation.adapters import repository
class AbstractUnitOfWork(abc.ABC):
products: repository.AbstractRepository
def __enter__(self) -> AbstractUnitOfWork:
return self
def __exit__(self,*args):
self.rollback()
@abc.abstractmethod
def commit(self):
raise NotImplementedError
@abc.abstractmethod
def rollback(self):
raise NotImplementedError
DEFAULT_SESSION_FACTORY = sessionmaker(bind=create_engine(
config.get_postgres_uri(),isolation_level="REPEATABLE READ",))
class sqlAlchemyUnitOfWork(AbstractUnitOfWork):
def __init__(self,session_factory=DEFAULT_SESSION_FACTORY):
self.session_factory = session_factory
def __enter__(self):
self.session = self.session_factory() # type: Session
self.products = repository.sqlAlchemyRepository(self.session)
return super().__enter__()
def __exit__(self,*args):
super().__exit__(*args)
self.session.close()
def commit(self):
self.session.commit()
def rollback(self):
self.session.rollback()
我正在尝试在 Flask 上测试我的端点,但我无法让它回滚每次测试后插入的数据。
为了解决我尝试安装包 pytest-flask-sqlalchemy
但出现以下错误的问题:
'sqlAlchemyUnitOfWork' object has no attribute 'engine'
我不太了解 pytest-flask-sqlalchemy
的工作原理,也不知道如何在测试后使工作单元回滚事务。
是否可以按照作者实现的方式使其工作?
已编辑
可以通过以下存储库复制我的情况:
https://github.com/Santana94/CosmicPythonRollbackTest
您应该通过克隆并运行 make all
来确认测试不会回滚之前的操作。
解决方法
最后,我必须在每次测试后都实现回滚功能。
当我看到一个名为 pytest-postgresql
的包在自身上实现它时,我开始工作了。我只是进行了调整以使测试回滚我正在使用的数据库数据。为此,我只需要在 conftest.py
上实现这个函数:
@pytest.fixture(scope='function')
def db_session():
engine = create_engine(config.get_postgres_uri(),echo=False,poolclass=NullPool)
metadata.create_all(engine)
pyramid_basemodel.Session = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
pyramid_basemodel.bind_engine(
engine,pyramid_basemodel.Session,should_create=True,should_drop=True)
yield pyramid_basemodel.Session
transaction.commit()
metadata.drop_all(engine)
之后,如果我想回滚事务,我必须将 db_session
作为测试的参数:
@pytest.mark.usefixtures('postgres_db')
@pytest.mark.usefixtures('restart_api')
def test_happy_path_returns_202_and_batch_is_allocated(db_session):
orderid = random_orderid()
sku,othersku = random_sku(),random_sku('other')
earlybatch = random_batchref(1)
laterbatch = random_batchref(2)
otherbatch = random_batchref(3)
api_client.post_to_add_batch(laterbatch,sku,100,'2011-01-02')
api_client.post_to_add_batch(earlybatch,'2011-01-01')
api_client.post_to_add_batch(otherbatch,othersku,None)
r = api_client.post_to_allocate(orderid,qty=3)
assert r.status_code == 202
r = api_client.get_allocation(orderid)
assert r.ok
assert r.json() == [
{'sku': sku,'batchref': earlybatch},]
可以在我的 GitHub 存储库上查看该实现的要求和其他方面。