无法在使用 Fixtures 的 Pytest 函数内实例化 Python 数据类(冻结)

Mat*_*ham 6 python sqlalchemy pytest python-dataclasses

我正在关注 Harry Percival 和 Bob Gregory 的《Python 中的架构模式》。

围绕第三(3)章介绍了SQLAlchemy的ORM测试。

一个需要session夹具的新测试,AttributeError, FrozenInstanceError由于cannot assign to field '_sa_instance_state'

需要注意的是,在创建 的实例时,其他测试不会失败OrderLine,但如果我只是将它们包含session到测试参数中,它们就会失败。

无论如何,我会直接进入代码。

conftest.py

@pytest.fixture
def local_db():
    engine = create_engine('sqlite:///:memory:')
    metadata.create_all(engine)
    return engine


@pytest.fixture
def session(local_db):
    start_mappers()
    yield sessionmaker(bind=local_db)()
    clear_mappers()
Run Code Online (Sandbox Code Playgroud)

模型.py

@dataclass(frozen=True)
class OrderLine:
    id: str
    sku: str
    quantity: int
Run Code Online (Sandbox Code Playgroud)

test_orm.py

def test_orderline_mapper_can_load_lines(session):
    session.execute(
        'INSERT INTO order_lines (order_id, sku, quantity) VALUES '
        '("order1", "RED-CHAIR", 12),'
        '("order1", "RED-TABLE", 13),'
        '("order2", "BLUE-LIPSTICK", 14)'
    )
    expected = [
        model.OrderLine("order1", "RED-CHAIR", 12),
        model.OrderLine("order1", "RED-TABLE", 13),
        model.OrderLine("order2", "BLUE-LIPSTICK", 14),
    ]
    assert session.query(model.OrderLine).all() == expected
Run Code Online (Sandbox Code Playgroud)

控制台错误pipenv run pytest test_orm.py

============================= test session starts =============================
platform linux -- Python 3.7.6, pytest-5.4.1, py-1.8.1, pluggy-0.13.1
rootdir: /home/[redacted]/Documents/architecture-patterns-python
collected 1 item                                                              

test_orm.py F                                                           [100%]

================================== FAILURES ===================================
____________________ test_orderline_mapper_can_load_lines _____________________

session = <sqlalchemy.orm.session.Session object at 0x7fd919ac5bd0>

    def test_orderline_mapper_can_load_lines(session):
        session.execute(
            'INSERT INTO order_lines (order_id, sku, quantity) VALUES '
            '("order1", "RED-CHAIR", 12),'
            '("order1", "RED-TABLE", 13),'
            '("order2", "BLUE-LIPSTICK", 14)'
        )
        expected = [
>           model.OrderLine("order1", "RED-CHAIR", 12),
            model.OrderLine("order1", "RED-TABLE", 13),
            model.OrderLine("order2", "BLUE-LIPSTICK", 14),
        ]

test_orm.py:13: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
<string>:2: in __init__
    ???
../../.local/share/virtualenvs/architecture-patterns-python-Qi2y0bev/lib64/python3.7/site-packages/sqlalchemy/orm/instrumentation.py:377: in _new_state_if_none
    self._state_setter(instance, state)
<string>:1: in set
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <[AttributeError("'OrderLine' object has no attribute '_sa_instance_state'") raised in repr()] OrderLine object at 0x7fd919a8cf50>
name = '_sa_instance_state'
value = <sqlalchemy.orm.state.InstanceState object at 0x7fd9198f7490>

>   ???
E   dataclasses.FrozenInstanceError: cannot assign to field '_sa_instance_state'

<string>:4: FrozenInstanceError
=========================== short test summary info ===========================
FAILED test_orm.py::test_orderline_mapper_can_load_lines - dataclasses.Froze...
============================== 1 failed in 0.06s ==============================
Run Code Online (Sandbox Code Playgroud)

其他问题

我了解覆盖的逻辑以及这些文件在做什么,但如果我缺乏基本的理解,请纠正我。

  1. conftest.py(用于所有 pytest 配置)正在设置一个session装置,它基本上在内存中设置一个临时数据库 - 使用 start 和 clear 映射器来确保 orm 模型定义绑定到 db 实例。
  2. model.py只是用于表示原子OrderLine对象的数据类。
  3. test_orm.pypytest 类来提供session夹具,以便为setup, execute, teardown运行测试的目的显式提供 db。

https://github.com/cosmicpython/code/issues/17提供的问题解决方案

Tom*_*Tom 4

SqlAlchemy 允许您覆盖在使用映射类和表时应用的一些属性检测。特别是以下内容允许 sqla 将状态保存在已检测的冻结数据类上。mapper这应该在调用关联数据类和 sql 表的函数之前应用。

from sqlalchemy.ext.instrumentation import InstrumentationManager

...

DEL_ATTR = object()


class FrozenDataclassInstrumentationManager(InstrumentationManager):
    def install_member(self, class_, key, implementation):
        self.originals.setdefault(key, class_.__dict__.get(key, DEL_ATTR))
        setattr(class_, key, implementation)

    def uninstall_member(self, class_, key):
        original = self.originals.pop(key, None)
        if original is not DEL_ATTR:
            setattr(class_, key, original)
        else:
            delattr(class_, key)

    def dispose(self, class_):
        del self.originals
        delattr(class_, "_sa_class_manager")
    
    def manager_getter(self, class_):
        def get(cls):
            return cls.__dict__["_sa_class_manager"]
        return get

    def manage(self, class_, manager):
        self.originals = {}
        setattr(class_, "_sa_class_manager", manager)

    def get_instance_dict(self, class_, instance):
        return instance.__dict__

    def install_state(self, class_, instance, state):
        instance.__dict__["state"] = state

    def remove_state(self, class_, instance, state):
        del instance.__dict__["state"]

    def state_getter(self, class_):
        def find(instance):
            return instance.__dict__["state"]
        return find




OrderLine.__sa_instrumentation_manager__ = FrozenDataclassInstrumentationManager
Run Code Online (Sandbox Code Playgroud)