ORM 2.0 style fake session mutations

[jonyscathe]

Is your feature request related to a problem? Please describe. Cannot currently read mock data back using session.execute(select(my_table)).all()

Describe the solution you'd like Currently the following works:

>>> session = UnifiedAlchemyMagicMock()
>>> session.add(Model(pk=1, foo='bar'))
>>> session.add(Model(pk=2, foo='baz'))
>>> session.query(Model).all()
[Model(foo='bar'), Model(foo='baz')]
>>> session.query(Model).get(1)
Model(foo='bar')
>>> session.query(Model).get(2)
Model(foo='baz')
>>> session.query(Model).get((2,))
Model(foo='baz')
>>> session.query(Model).get({"pk2" : 2}))
Model(foo='baz')

I would like the following to also work:

>>> session = UnifiedAlchemyMagicMock()
>>> session.add(Model(pk=1, foo='bar'))
>>> session.add(Model(pk=2, foo='baz'))
>>> session.execute(select(Model)).all()
[Model(foo='bar'), Model(foo='baz')]
>>> session.execute(select(Model)).get(1)
Model(foo='bar')
>>> session.execute(select(Model)).get(2)
Model(foo='baz')
>>> session.execute(select(Model)).get((2,))
Model(foo='baz')
>>> session.execute(select(Model)).get({"pk2" : 2}))
Model(foo='baz')

Describe alternatives you've considered Can stick with ORM 1.0 style queries in tests if this becomes too difficult.

Additional context I have tested the following change locally and it does work (for at least very simple cases).

Changing in mocking.py:

        if _mock_name == "add":
            to_add = args[0]
            query_call = mock.call.query(type(to_add))

            mocked_data = next(iter(filter(lambda i: i[0] == [query_call], _mock_data)), None)
            if mocked_data:
                mocked_data[1].append(to_add)
            else:
                _mock_data.append(([query_call], [to_add]))

to:

        if _mock_name == "add":
            to_add = args[0]
            query_call = mock.call.query(type(to_add))

            query_mocked_data = next(iter(filter(lambda i: i[0] == [query_call], _mock_data)), None)
            if query_mocked_data:
                query_mocked_data[1].append(to_add)
            else:
                _mock_data.append(([query_call], [to_add]))

            execute_call = mock.call.execute(select(type(to_add)))

            execute_mocked_data = next(iter(filter(lambda i: i[0] == [execute_call], _mock_data)), None)
            if execute_mocked_data:
                execute_mocked_data[1].append(to_add)
            else:
                _mock_data.append(([execute_call], [to_add]))

plus adding from sqlalchemy import select up the top.

After that, when I have done a simple add I can read back my data using either session.execute(select(my_table)).all() or `session.query(my_table).all() and that all works fine.

If it was that simple I would have a PR up already. However, that code would likely break things for some people. Firstly, I don't believe using select() in that way is not allowable in older versions of sqlalchemy (such as 1.3.22). In the current mock-alchemy tests, the 1.3.22 tests fail with a TypeError: 'DeclarativeMeta' object is not iterable. Secondly, if the 'my_table' object is being mocked and isn't a Column expression, FROM clause or other columns clause element then the select(type(to_add)) will throw an sqlalchemy.exc.ArgumentError.

Both of these issues could be resolved with a simple try/except block:

        try:
            execute_call = mock.call.execute(select(type(to_add)))

            execute_mocked_data = next(iter(filter(lambda i: i[0] == [execute_call], _mock_data)), None)
            if execute_mocked_data:
                execute_mocked_data[1].append(to_add)
            else:
                _mock_data.append(([execute_call], [to_add]))
        except (TypeError, ArgumentError):
            # TypeError indicates an old version of sqlalcemy that does not support executing select(table) statements.
            # ArgumentError indicates a mocked table that cannot be selected, so cannot be mocked this way.
            pass

But before doing something like that I wanted to get your opinion on that style of programming as some people are not huge fans of those sort of try/except blocks.

[jonyscathe]

Already realised a slight issue. The:

execute_mocked_data = next(iter(filter(lambda i: i[0] == [execute_call], _mock_data)), None)

Actually needs to be

execute_mocked_data = next(iter(filter(lambda i: i[0] == [ExpressionMatcher(execute_call)], _mock_data)), None)

This is because each select object is a different object and we need to be able to actually compare them and have them match in this statement.

[jonyscathe]

And FYI, after changing this in a local branch, the following quick test (copied and modified from test_update_calls()) passes with sqlalchemy >= 1.4.0

def test_update_calls_orm_2() -> None:
    """Tests that update calls have no impact on UnifiedAlchemyMagicMock sessions.

    This should eventually work the same for any unknown calls.
    """
    expected_row = [Model(pk1="1234", name="test")]
    mock_session = UnifiedAlchemyMagicMock(
        data=[
            (
                [
                    mock.call.execute(select(Model).where(Model.pk1 == 3)),
                ],
                expected_row,
            )
        ]
    )
    # Test all()
    actual_row = mock_session.execute(select(Model).where(Model.pk1 == 3)).all()
    assert expected_row == actual_row
    mock_session.execute(select(Model).where(Model.pk1 == 3)).update({"pk1": 3})
    actual_row = mock_session.execute(select(Model).where(Model.pk1 == 3)).all()
    assert expected_row == actual_row
    # Test delete()
    assert None is mock_session.execute(select(Model).where(Model.pk1 == 3)).update(
        {"pk1": 3}, synchronize_session="evaluate"
    )
    deleted_count = mock_session.execute(select(Model).where(Model.pk1 == 3)).delete()
    assert 1 == deleted_count
    actual_row = mock_session.execute(select(Model).where(Model.pk1 == 3)).all()
    assert [] == actual_row

And I do have a branch with this change in as it currently stands: https://github.com/jonyscathe/mock-alchemy/tree/jonyscathe/execute_mutate

But figured I would for a comment or two before writing any tests and making a PR

[jonyscathe]

Actually... I might work on this a bit more.

Ideally, things like

session.execute(insert(Model), [{'pk': 1, 'foo': 'bar'}])
session.execute(insert(Model).returning(Model), [{'pk': 2, 'foo': 'bar'}]).scalar()

Would also work for adding data. While I do like a lot of the sqlalchemy 2.0 changes, it does complicate things a bit for mock-alchemy...

Edit: Have these working in my branch now as well as:

session.execute(insert(Model).values([{'pk': 1, 'foo': 'bar'}]))

Treating execute as both a mutate and unify keyword - ideally will be able to work out when it should/shouldn't unify - for example with the two statements at the top of this post, the first should not unify, the second should - the key difference being the .returning() addition.

Code isn't still WIP, not linted, tested, etc. Still working on update/delete and then when to/not unify. I also don't have session.execute(insert(Model).returning(Model), ({'pk': 2, 'foo': 'bar')}).scalar() actually returning the data for now... That one is a little trickier. The insert(X).returning can basically be substituted for select however the where or filter clause might be a tad trickier given that this can be used to insert a large number of rows in one go.