python - 使用 SQLAlchemy 1.4 经典/命令式映射样式？

Question

我在我的 Python 项目中使用干净的架构和 TDD 开发方法。从 SQLAlchemy 1.3 更新到 SQLAlchemy 1.4 后，无法针对内存中的 Postgres 数据库进行测试，我找不到解决该问题的方法。

遵循 DDD 原则，该项目使用新的命令式映射语法，替换经典映射声明。

这是一个最小的(非)工作示例，改编自 SQLAlchemy 文档: https://docs.sqlalchemy.org/en/14/orm/mapping_styles.html#orm-imperative-mapping

它需要在本地安装和运行 PostgreSQL。

我的应用程序/orm.py

from sqlalchemy import MetaData, Table, Column, Integer, String
from sqlalchemy.orm import registry
from myapp import model

mapper_registry = registry()
metadata = MetaData()

user_table = Table(
    'tb_user',
    mapper_registry.metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('fullname', String(50)),
    Column('nickname', String(12))
)

mapper_registry.map_imperatively(model.User, user_table)

我的应用程序/模型.py

from dataclasses import dataclass
from dataclasses import field

@dataclass
class User:
    id: int = field(init=False)
    name: str = ""
    fullname: str = ""
    nickname: str = ""

测试/test_postgresql_inmemory.py

import tempfile

import pytest
from pytest_postgresql import factories
from sqlalchemy import create_engine
from sqlalchemy import text
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import configure_mappers
from myapp import model
from myapp.orm import mapper_registry
from sqlalchemy.orm import sessionmaker



# here, we set up postgresql in-memory:

socket_dir = tempfile.TemporaryDirectory()
postgresql_my_proc = factories.postgresql_proc(
    port=None,
    unixsocketdir=socket_dir.name,
)
postgresql_my = factories.postgresql("postgresql_my_proc")


@pytest.fixture
def in_memory_db(postgresql_my):
    def db_creator():
        return postgresql_my.cursor().connection

    engine = create_engine("postgresql+psycopg2://", creator=db_creator)
    mapper_registry.metadata.create_all(bind=engine)
    return engine


@pytest.fixture
def session(in_memory_db):
    clear_mappers()
    configure_mappers()
    Session = sessionmaker(bind=in_memory_db)
    session = Session()
    yield session
    clear_mappers()



def test_User_mapper_can_add(session):
    user = model.User(fullname="John Smith")
    session.add(user)
    session.commit()
    rows = list(session.execute("SELECT fullname FROM tb_user"))
    assert rows == [("John Smith",)]

结果

===== test session starts =====
platform linux -- Python 3.9.4, pytest-5.4.3, py-1.10.0, pluggy-0.13.1 
rootdir: /home/lionel/code/sqla14_test
plugins: postgresql-3.1.1 
collected 1 item 
tests/test_postgresql_inmemory.py F                                                                            [100%] 
==== FAILURES =====
___ test_User_mapper_can_add ___
self = <sqlalchemy.orm.session.Session object at 0x7fe876060c70> 
instance = <[AttributeError("'User' object has no attribute 'id'") raised in repr()] User object at 0x7fe875f487c0> 
_warn = True 
    def add(self, instance, _warn=True): 
        """Place an object in the ``Session``. 
        Its state will be persisted to the database on the next flush 
        operation. 
        Repeated calls to ``add()`` will be ignored. The opposite of ``add()`` 
        is ``expunge()``. 
        """ 
        if _warn and self._warn_on_events: 
            self._flush_warning("Session.add()") 
        try: 
>           state = attributes.instance_state(instance) 
E           AttributeError: 'User' object has no attribute '_sa_instance_state' 
../../myvenv/lib/python3.9/site-packages/sqlalchemy/orm/session.py:2554: AttributeError 
The above exception was the direct cause of the following exception: 
session = <sqlalchemy.orm.session.Session object at 0x7fe876060c70> 
    def test_User_mapper_can_add(session): 
        user = model.User(fullname="John Smith") 
>       session.add(user) 
tests/test_postgresql_inmemory.py:53: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../.cache/pypoetry/virtualenvs/sqla14-5BJjO56U-py3.9/lib/python3.9/site-packages/sqlalchemy/orm/session.py:2556: in add 
    util.raise_( 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
    def raise_( 
        exception, with_traceback=None, replace_context=None, from_=False 
    ): 
        r"""implement "raise" with cause support. 
        :param exception: exception to raise 
        :param with_traceback: will call exception.with_traceback() 
        :param replace_context: an as-yet-unsupported feature.  This is 
         an exception object which we are "replacing", e.g., it's our 
         "cause" but we don't want it printed.    Basically just what 
         ``__suppress_context__`` does but we don't want to suppress 
         the enclosing context, if any.  So for now we make it the 
         cause. 
        :param from\_: the cause.  this actually sets the cause and doesn't 
         hope to hide it someday. 
        """ 
        if with_traceback is not None: 
            exception = exception.with_traceback(with_traceback) 
        if from_ is not False: 
            exception.__cause__ = from_ 
        elif replace_context is not None: 
            # no good solution here, we would like to have the exception 
            # have only the context of replace_context.__context__ so that the 
            # intermediary exception does not change, but we can't figure 
            # that out. 
            exception.__cause__ = replace_context 
        try: 
>           raise exception 
E           sqlalchemy.orm.exc.UnmappedInstanceError: Class 'myapp.model.User' is not mapped 
../../myvenv/lib/python3.9/site-packages/sqlalchemy/util/compat.py:207: UnmappedInstanceError 
-------- Captured stderr setup ----
sh: warning: setlocale: LC_ALL: cannot change locale (C.UTF-8) 
/bin/sh: warning: setlocale: LC_ALL: cannot change locale (C.UTF-8) 
========short test summary info =====
FAILED tests/test_postgresql_inmemory.py::test_User_mapper_can_add - sqlalchemy.orm.exc.UnmappedInstanceError: Clas...
=========1 failed in 1.18s ==========

您是否看到需要更改哪些内容才能通过测试？

Answer 1

我可以通过包装 mapper_registry.map_imperatively(...) 使测试通过在 start_mappers 函数中，就像我之前做的那样。

我最初认为我必须用 configure_mappers 替换它 - SQLAlchemy documentation .

我的应用程序/orm.py

from sqlalchemy import MetaData, Table, Column, Integer, String
from sqlalchemy.orm import registry
from myapp import model

metadata = MetaData()
mapper_registry = registry(metadata=metadata)

user_table = Table(
    'tb_user',
    mapper_registry.metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('fullname', String(50)),
    Column('nickname', String(12))
)

def start_mappers():
    mapper_registry.map_imperatively(model.User, user_table)

测试/test_postgresql_inmemory.py

摘录:

from myapp.orm import start_mappers

# …

@pytest.fixture
def session(in_memory_db):
    clear_mappers()
    start_mappers()
    Session = sessionmaker(bind=in_memory_db)
    session = Session()
    yield session
    clear_mappers()

def test_User_mapper_can_add(session):
    user = model.User(fullname="John Smith")
    session.add(user)
    session.commit()
    rows = list(session.execute("SELECT fullname FROM tb_user"))
    assert rows == [("John Smith", )]

结果

[Sat Jun  5 19:42:46 2021] Running: py.test tests/test_postgresql_inmemory.py
===== test session starts ====
platform linux -- Python 3.9.4, pytest-5.4.3, py-1.10.0, pluggy-0.13.1
rootdir: /home/lionel/code/myapp
plugins: postgresql-3.1.1
collected 1 item

tests/test_postgresql_inmemory.py .          [100%]

===== 1 passed in 1.12s ====