I am using the clean architecture & TDD development method in my Python project. Updating from SQLAlchemy 1.3 to SQLAlchemy 1.4 broke the ability to test against an in-memory Postgres DB, and I can’t find how to fix the problem.
Following DDD principles, the project uses the new imperative mapping syntax which replace classical mapping declarations.
Here is a minimal (non)-working example, adapted from SQLAlchemy documentation: https://docs.sqlalchemy.org/en/14/orm/mapping_styles.html#orm-imperative-mapping
It requires installing and run PostgreSQL locally.
myapp/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)
myapp/model.py
from dataclasses import dataclass
from dataclasses import field
@dataclass
class User:
id: int = field(init=False)
name: str = ""
fullname: str = ""
nickname: str = ""
tests/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",)]
Result
===== 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 ==========
Do you see what needs to be changed to make the test pass?
Advertisement
Answer
I could make the test pass by wrapping the mapper_registry.map_imperatively(…)
in a start_mappers function, like I was doing before.
I initially thought that I had to replace this by configure_mappers – SQLAlchemy documentation.
myapp/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)
tests/test_postgresql_inmemory.py
extract:
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", )]
result
[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 ====