Method test_batch_interaction

test/orm/test_unitofworkv2.py:2068–2160

test batching groups same-structured, primary key present statements together.

(self)

        )

    def test_batch_interaction(self):
        """test batching groups same-structured, primary
        key present statements together.

        """

        t = self.tables.t

        class T(ComparableEntity):
            pass

        mp = self.mapper_registry.map_imperatively(T, t)
        sess = fixture_session()
        sess.add_all(
            [
                T(data="t1"),
                T(data="t2"),
                T(id=3, data="t3"),
                T(id=4, data="t4"),
                T(id=5, data="t5"),
                T(id=6, data=func.lower("t6")),
                T(id=7, data="t7"),
                T(id=8, data="t8"),
                T(id=9, data="t9", def_="def2"),
                T(id=10, data="t10", def_="def3"),
                T(id=11, data="t11"),
            ]
        )

        eager_defaults = mp._prefer_eager_defaults(
            testing.db.dialect, mp.local_table
        )

        if eager_defaults:
            tdef_col = ", t.def_"
            tdef_returning = " RETURNING t.def_"
        else:
            tdef_col = ""
            tdef_returning = ""

        self.assert_sql_execution(
            testing.db,
            sess.flush,
            Conditional(
                testing.db.dialect.insert_executemany_returning,
                [
                    CompiledSQL(
                        "INSERT INTO t (data) VALUES (:data) "
                        f"RETURNING t.id{tdef_col}",
                        [{"data": "t1"}, {"data": "t2"}],
                    ),
                ],
                [
                    CompiledSQL(
                        "INSERT INTO t (data) VALUES (:data)", {"data": "t1"}
                    ),
                    CompiledSQL(
                        "INSERT INTO t (data) VALUES (:data)", {"data": "t2"}