from sqlalchemy import FetchedValue from sqlalchemy import ForeignKey from sqlalchemy import Identity from sqlalchemy import insert from sqlalchemy import inspect from sqlalchemy import Integer from sqlalchemy import String from sqlalchemy import testing from sqlalchemy import update from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures from sqlalchemy.testing import mock from sqlalchemy.testing.assertsql import CompiledSQL from sqlalchemy.testing.assertsql import Conditional from sqlalchemy.testing.fixtures import fixture_session from sqlalchemy.testing.schema import Column from sqlalchemy.testing.schema import Table from test.orm import _fixtures class BulkTest(testing.AssertsExecutionResults): run_inserts = None run_define_tables = "each" class BulkInsertUpdateVersionId(BulkTest, fixtures.MappedTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "version_table", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("version_id", Integer, nullable=False), Column("value", String(40), nullable=False), ) @classmethod def setup_classes(cls): class Foo(cls.Comparable): pass @classmethod def setup_mappers(cls): Foo, version_table = cls.classes.Foo, cls.tables.version_table cls.mapper_registry.map_imperatively( Foo, version_table, version_id_col=version_table.c.version_id ) @testing.emits_warning(r".*versioning cannot be verified") def test_bulk_insert_via_save(self): Foo = self.classes.Foo s = fixture_session() s.bulk_save_objects([Foo(value="value")]) eq_(s.query(Foo).all(), [Foo(version_id=1, value="value")]) @testing.emits_warning(r".*versioning cannot be verified") def test_bulk_update_via_save(self): Foo = self.classes.Foo s = fixture_session() s.add(Foo(value="value")) s.commit() f1 = s.query(Foo).first() f1.value = "new value" s.bulk_save_objects([f1]) s.expunge_all() eq_(s.query(Foo).all(), [Foo(version_id=2, value="new value")]) class BulkInsertUpdateTest(BulkTest, _fixtures.FixtureTest): __backend__ = True @classmethod def setup_mappers(cls): User, Address, Order = cls.classes("User", "Address", "Order") u, a, o = cls.tables("users", "addresses", "orders") cls.mapper_registry.map_imperatively(User, u) cls.mapper_registry.map_imperatively(Address, a) cls.mapper_registry.map_imperatively(Order, o) @testing.combinations( "save_objects", "insert_mappings", "insert_stmt", argnames="statement_type", ) @testing.variation("return_defaults", [True, False]) def test_bulk_save_return_defaults(self, statement_type, return_defaults): (User,) = self.classes("User") s = fixture_session() if statement_type == "save_objects": objects = [User(name="u1"), User(name="u2"), User(name="u3")] assert "id" not in objects[0].__dict__ returning_users_id = " RETURNING users.id" with self.sql_execution_asserter() as asserter: s.bulk_save_objects(objects, return_defaults=return_defaults) elif statement_type == "insert_mappings": data = [dict(name="u1"), dict(name="u2"), dict(name="u3")] returning_users_id = " RETURNING users.id" with self.sql_execution_asserter() as asserter: s.bulk_insert_mappings( User, data, return_defaults=return_defaults ) elif statement_type == "insert_stmt": data = [dict(name="u1"), dict(name="u2"), dict(name="u3")] # for statement, "return_defaults" is heuristic on if we are # a joined inh mapping if we don't otherwise include # .returning() on the statement itself returning_users_id = "" with self.sql_execution_asserter() as asserter: s.execute(insert(User), data) asserter.assert_( Conditional( ( return_defaults and testing.db.dialect.insert_executemany_returning ) or statement_type == "insert_stmt", [ CompiledSQL( "INSERT INTO users (name) " f"VALUES (:name){returning_users_id}", [{"name": "u1"}, {"name": "u2"}, {"name": "u3"}], ), ], [ Conditional( return_defaults, [ CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "u1"}], ), CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "u2"}], ), CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "u3"}], ), ], [ CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [ {"name": "u1"}, {"name": "u2"}, {"name": "u3"}, ], ), ], ) ], ) ) if statement_type == "save_objects": if return_defaults: eq_(objects[0].__dict__["id"], 1) eq_(inspect(objects[0]).key, (User, (1,), None)) else: assert "id" not in objects[0].__dict__ eq_(inspect(objects[0]).key, None) elif statement_type == "insert_mappings": # test for #11661 if return_defaults: eq_(data[0]["id"], 1) else: assert "id" not in data[0] def test_bulk_save_objects_defaults_key(self): User = self.classes.User pes = [User(name=f"foo{i}") for i in range(3)] s = fixture_session() s.bulk_save_objects(pes, return_defaults=True) key = inspect(pes[0]).key s.commit() eq_(inspect(s.get(User, 1)).key, key) def test_bulk_save_mappings_preserve_order(self): (User,) = self.classes("User") s = fixture_session() # commit some object into db user1 = User(name="i1") user2 = User(name="i2") s.add(user1) s.add(user2) s.commit() # make some changes user1.name = "u1" user3 = User(name="i3") s.add(user3) user2.name = "u2" objects = [user1, user3, user2] from sqlalchemy import inspect def _bulk_save_mappings( mapper, mappings, isupdate, isstates, return_defaults, update_changed_only, render_nulls, ): mock_method(list(mappings), isupdate) mock_method = mock.Mock() with mock.patch.object(s, "_bulk_save_mappings", _bulk_save_mappings): s.bulk_save_objects(objects) eq_( mock_method.mock_calls, [ mock.call([inspect(user1)], True), mock.call([inspect(user3)], False), mock.call([inspect(user2)], True), ], ) mock_method = mock.Mock() with mock.patch.object(s, "_bulk_save_mappings", _bulk_save_mappings): s.bulk_save_objects(objects, preserve_order=False) eq_( mock_method.mock_calls, [ mock.call([inspect(user3)], False), mock.call([inspect(user1), inspect(user2)], True), ], ) def test_bulk_save_no_defaults(self): (User,) = self.classes("User") s = fixture_session() objects = [User(name="u1"), User(name="u2"), User(name="u3")] assert "id" not in objects[0].__dict__ with self.sql_execution_asserter() as asserter: s.bulk_save_objects(objects) asserter.assert_( CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "u1"}, {"name": "u2"}, {"name": "u3"}], ) ) assert "id" not in objects[0].__dict__ def test_bulk_save_updated_include_unchanged(self): (User,) = self.classes("User") s = fixture_session(expire_on_commit=False) objects = [User(name="u1"), User(name="u2"), User(name="u3")] s.add_all(objects) s.commit() objects[0].name = "u1new" objects[2].name = "u3new" s = fixture_session() with self.sql_execution_asserter() as asserter: s.bulk_save_objects(objects, update_changed_only=False) asserter.assert_( CompiledSQL( "UPDATE users SET name=:name WHERE users.id = :users_id", [ {"users_id": 1, "name": "u1new"}, {"users_id": 2, "name": "u2"}, {"users_id": 3, "name": "u3new"}, ], ) ) @testing.combinations("update_mappings", "update_stmt") def test_bulk_update(self, statement_type): User = self.classes.User s = fixture_session(expire_on_commit=False) objects = [User(name="u1"), User(name="u2"), User(name="u3")] s.add_all(objects) s.commit() s = fixture_session() data = [ {"id": 1, "name": "u1new"}, {"id": 2, "name": "u2"}, {"id": 3, "name": "u3new"}, ] if statement_type == "update_mappings": with self.sql_execution_asserter() as asserter: s.bulk_update_mappings(User, data) elif statement_type == "update_stmt": with self.sql_execution_asserter() as asserter: s.execute(update(User), data) asserter.assert_( CompiledSQL( "UPDATE users SET name=:name WHERE users.id = :users_id", [ {"users_id": 1, "name": "u1new"}, {"users_id": 2, "name": "u2"}, {"users_id": 3, "name": "u3new"}, ], ) ) def test_bulk_insert(self): (User,) = self.classes("User") s = fixture_session() with self.sql_execution_asserter() as asserter: s.bulk_insert_mappings( User, [ {"id": 1, "name": "u1new"}, {"id": 2, "name": "u2"}, {"id": 3, "name": "u3new"}, ], ) asserter.assert_( CompiledSQL( "INSERT INTO users (id, name) VALUES (:id, :name)", [ {"id": 1, "name": "u1new"}, {"id": 2, "name": "u2"}, {"id": 3, "name": "u3new"}, ], ) ) def test_bulk_insert_render_nulls(self): (Order,) = self.classes("Order") s = fixture_session() with self.sql_execution_asserter() as asserter: s.bulk_insert_mappings( Order, [ {"id": 1, "description": "u1new"}, {"id": 2, "description": None}, {"id": 3, "description": "u3new"}, ], render_nulls=True, ) asserter.assert_( CompiledSQL( "INSERT INTO orders (id, description) " "VALUES (:id, :description)", [ {"id": 1, "description": "u1new"}, {"id": 2, "description": None}, {"id": 3, "description": "u3new"}, ], ) ) class BulkUDPostfetchTest(BulkTest, fixtures.MappedTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "a", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("x", Integer), Column( "y", Integer, server_default=FetchedValue(), server_onupdate=FetchedValue(), ), ) @classmethod def setup_classes(cls): class A(cls.Comparable): pass @classmethod def setup_mappers(cls): A = cls.classes.A a = cls.tables.a cls.mapper_registry.map_imperatively(A, a) def test_insert_w_fetch(self): A = self.classes.A s = fixture_session() a1 = A(x=1) s.bulk_save_objects([a1]) s.commit() def test_update_w_fetch(self): A = self.classes.A s = fixture_session() a1 = A(x=1, y=2) s.add(a1) s.commit() eq_(a1.id, 1) # force a load a1.x = 5 s.expire(a1, ["y"]) assert "y" not in a1.__dict__ s.bulk_save_objects([a1]) s.commit() eq_(a1.x, 5) eq_(a1.y, 2) class BulkUDTestAltColKeys(BulkTest, fixtures.MappedTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "people_keys", metadata, Column("person_id", Integer, primary_key=True, key="id"), Column("name", String(50), key="personname"), ) Table( "people_attrs", metadata, Column("person_id", Integer, primary_key=True), Column("name", String(50)), ) Table( "people_both", metadata, Column("person_id", Integer, primary_key=True, key="id_key"), Column("name", String(50), key="name_key"), ) @classmethod def setup_classes(cls): class PersonKeys(cls.Comparable): pass class PersonAttrs(cls.Comparable): pass class PersonBoth(cls.Comparable): pass @classmethod def setup_mappers(cls): PersonKeys, PersonAttrs, PersonBoth = cls.classes( "PersonKeys", "PersonAttrs", "PersonBoth" ) people_keys, people_attrs, people_both = cls.tables( "people_keys", "people_attrs", "people_both" ) cls.mapper_registry.map_imperatively(PersonKeys, people_keys) cls.mapper_registry.map_imperatively( PersonAttrs, people_attrs, properties={ "id": people_attrs.c.person_id, "personname": people_attrs.c.name, }, ) cls.mapper_registry.map_imperatively( PersonBoth, people_both, properties={ "id": people_both.c.id_key, "personname": people_both.c.name_key, }, ) def test_insert_keys(self): asserter = self._test_insert(self.classes.PersonKeys) asserter.assert_( CompiledSQL( "INSERT INTO people_keys (person_id, name) " "VALUES (:id, :personname)", [{"id": 5, "personname": "thename"}], ) ) def test_insert_attrs(self): asserter = self._test_insert(self.classes.PersonAttrs) asserter.assert_( CompiledSQL( "INSERT INTO people_attrs (person_id, name) " "VALUES (:person_id, :name)", [{"person_id": 5, "name": "thename"}], ) ) def test_insert_both(self): asserter = self._test_insert(self.classes.PersonBoth) asserter.assert_( CompiledSQL( "INSERT INTO people_both (person_id, name) " "VALUES (:id_key, :name_key)", [{"id_key": 5, "name_key": "thename"}], ) ) @testing.combinations( ("states",), ("dicts",), ) def test_update_keys(self, type_): if type_ == "states": asserter = self._test_update_states(self.classes.PersonKeys) else: asserter = self._test_update(self.classes.PersonKeys) asserter.assert_( CompiledSQL( "UPDATE people_keys SET name=:personname " "WHERE people_keys.person_id = :people_keys_person_id", [{"personname": "newname", "people_keys_person_id": 5}], ) ) @testing.combinations( ("states",), ("dicts",), ) @testing.requires.updateable_autoincrement_pks def test_update_attrs(self, type_): if type_ == "states": asserter = self._test_update_states(self.classes.PersonAttrs) else: asserter = self._test_update(self.classes.PersonAttrs) asserter.assert_( CompiledSQL( "UPDATE people_attrs SET name=:name " "WHERE people_attrs.person_id = :people_attrs_person_id", [{"name": "newname", "people_attrs_person_id": 5}], ) ) @testing.requires.updateable_autoincrement_pks def test_update_both(self): # want to make sure that before [ticket:3849], this did not have # a successful behavior or workaround asserter = self._test_update(self.classes.PersonBoth) asserter.assert_( CompiledSQL( "UPDATE people_both SET name=:name_key " "WHERE people_both.person_id = :people_both_person_id", [{"name_key": "newname", "people_both_person_id": 5}], ) ) def _test_insert(self, person_cls): Person = person_cls s = fixture_session() with self.sql_execution_asserter(testing.db) as asserter: s.bulk_insert_mappings( Person, [{"id": 5, "personname": "thename"}] ) eq_(s.query(Person).first(), Person(id=5, personname="thename")) return asserter def _test_update(self, person_cls): Person = person_cls s = fixture_session() s.add(Person(id=5, personname="thename")) s.commit() with self.sql_execution_asserter(testing.db) as asserter: s.bulk_update_mappings( Person, [{"id": 5, "personname": "newname"}] ) eq_(s.query(Person).first(), Person(id=5, personname="newname")) return asserter def _test_update_states(self, person_cls): Person = person_cls s = fixture_session() s.add(Person(id=5, personname="thename")) s.commit() p = s.get(Person, 5) with self.sql_execution_asserter(testing.db) as asserter: p.personname = "newname" s.bulk_save_objects([p]) eq_(s.query(Person).first(), Person(id=5, personname="newname")) return asserter class BulkInheritanceTest(BulkTest, fixtures.MappedTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "people", metadata, Column( "person_id", Integer, primary_key=True, test_needs_autoincrement=True, ), Column("name", String(50)), Column("type", String(30)), ) Table( "engineers", metadata, Column( "person_id", Integer, ForeignKey("people.person_id"), primary_key=True, ), Column("status", String(30)), Column("primary_language", String(50)), ) Table( "managers", metadata, Column( "person_id", Integer, ForeignKey("people.person_id"), primary_key=True, ), Column("status", String(30)), Column("manager_name", String(50)), ) Table( "boss", metadata, Column( "boss_id", Integer, ForeignKey("managers.person_id"), primary_key=True, ), Column("golf_swing", String(30)), ) @classmethod def setup_classes(cls): class Base(cls.Comparable): pass class Person(Base): pass class Engineer(Person): pass class Manager(Person): pass class Boss(Manager): pass @classmethod def setup_mappers(cls): Person, Engineer, Manager, Boss = cls.classes( "Person", "Engineer", "Manager", "Boss" ) p, e, m, b = cls.tables("people", "engineers", "managers", "boss") cls.mapper_registry.map_imperatively( Person, p, polymorphic_on=p.c.type, polymorphic_identity="person" ) cls.mapper_registry.map_imperatively( Engineer, e, inherits=Person, polymorphic_identity="engineer" ) cls.mapper_registry.map_imperatively( Manager, m, inherits=Person, polymorphic_identity="manager" ) cls.mapper_registry.map_imperatively( Boss, b, inherits=Manager, polymorphic_identity="boss" ) def test_bulk_save_joined_inh_return_defaults(self): Person, Engineer, Manager, Boss = self.classes( "Person", "Engineer", "Manager", "Boss" ) s = fixture_session() objects = [ Manager(name="m1", status="s1", manager_name="mn1"), Engineer(name="e1", status="s2", primary_language="l1"), Engineer(name="e2", status="s3", primary_language="l2"), Boss(name="b1", status="s3", manager_name="mn2", golf_swing="g1"), ] assert "person_id" not in objects[0].__dict__ with self.sql_execution_asserter() as asserter: s.bulk_save_objects(objects, return_defaults=True) asserter.assert_( CompiledSQL( "INSERT INTO people (name, type) VALUES (:name, :type)", [{"type": "manager", "name": "m1"}], ), CompiledSQL( "INSERT INTO managers (person_id, status, manager_name) " "VALUES (:person_id, :status, :manager_name)", [{"person_id": 1, "status": "s1", "manager_name": "mn1"}], ), Conditional( testing.db.dialect.insert_executemany_returning, [ CompiledSQL( "INSERT INTO people (name, type) " "VALUES (:name, :type) RETURNING people.person_id", [ {"type": "engineer", "name": "e1"}, {"type": "engineer", "name": "e2"}, ], ), ], [ CompiledSQL( "INSERT INTO people (name, type) " "VALUES (:name, :type)", [{"type": "engineer", "name": "e1"}], ), CompiledSQL( "INSERT INTO people (name, type) " "VALUES (:name, :type)", [{"type": "engineer", "name": "e2"}], ), ], ), CompiledSQL( "INSERT INTO engineers (person_id, status, primary_language) " "VALUES (:person_id, :status, :primary_language)", [ {"person_id": 2, "status": "s2", "primary_language": "l1"}, {"person_id": 3, "status": "s3", "primary_language": "l2"}, ], ), CompiledSQL( "INSERT INTO people (name, type) VALUES (:name, :type)", [{"type": "boss", "name": "b1"}], ), CompiledSQL( "INSERT INTO managers (person_id, status, manager_name) " "VALUES (:person_id, :status, :manager_name)", [{"person_id": 4, "status": "s3", "manager_name": "mn2"}], ), CompiledSQL( "INSERT INTO boss (boss_id, golf_swing) VALUES " "(:boss_id, :golf_swing)", [{"boss_id": 4, "golf_swing": "g1"}], ), ) eq_(objects[0].__dict__["person_id"], 1) eq_(objects[3].__dict__["person_id"], 4) eq_(objects[3].__dict__["boss_id"], 4) def test_bulk_save_joined_inh_no_defaults(self): Person, Engineer, Manager, Boss = self.classes( "Person", "Engineer", "Manager", "Boss" ) s = fixture_session() with self.sql_execution_asserter() as asserter: s.bulk_save_objects( [ Manager( person_id=1, name="m1", status="s1", manager_name="mn1" ), Engineer( person_id=2, name="e1", status="s2", primary_language="l1", ), Engineer( person_id=3, name="e2", status="s3", primary_language="l2", ), Boss( person_id=4, boss_id=4, name="b1", status="s3", manager_name="mn2", golf_swing="g1", ), ] ) # the only difference here is that common classes are grouped together. # at the moment it doesn't lump all the "people" tables from # different classes together. asserter.assert_( CompiledSQL( "INSERT INTO people (person_id, name, type) VALUES " "(:person_id, :name, :type)", [{"person_id": 1, "type": "manager", "name": "m1"}], ), CompiledSQL( "INSERT INTO managers (person_id, status, manager_name) " "VALUES (:person_id, :status, :manager_name)", [{"status": "s1", "person_id": 1, "manager_name": "mn1"}], ), CompiledSQL( "INSERT INTO people (person_id, name, type) VALUES " "(:person_id, :name, :type)", [ {"person_id": 2, "type": "engineer", "name": "e1"}, {"person_id": 3, "type": "engineer", "name": "e2"}, ], ), CompiledSQL( "INSERT INTO engineers (person_id, status, primary_language) " "VALUES (:person_id, :status, :primary_language)", [ {"person_id": 2, "status": "s2", "primary_language": "l1"}, {"person_id": 3, "status": "s3", "primary_language": "l2"}, ], ), CompiledSQL( "INSERT INTO people (person_id, name, type) VALUES " "(:person_id, :name, :type)", [{"person_id": 4, "type": "boss", "name": "b1"}], ), CompiledSQL( "INSERT INTO managers (person_id, status, manager_name) " "VALUES (:person_id, :status, :manager_name)", [{"status": "s3", "person_id": 4, "manager_name": "mn2"}], ), CompiledSQL( "INSERT INTO boss (boss_id, golf_swing) VALUES " "(:boss_id, :golf_swing)", [{"boss_id": 4, "golf_swing": "g1"}], ), ) @testing.combinations("insert_mappings", "insert_stmt") def test_bulk_insert_joined_inh_return_defaults(self, statement_type): Person, Engineer, Manager, Boss = self.classes( "Person", "Engineer", "Manager", "Boss" ) s = fixture_session() data = [ dict( name="b1", status="s1", manager_name="mn1", golf_swing="g1", ), dict( name="b2", status="s2", manager_name="mn2", golf_swing="g2", ), dict( name="b3", status="s3", manager_name="mn3", golf_swing="g3", ), ] if statement_type == "insert_mappings": with self.sql_execution_asserter() as asserter: s.bulk_insert_mappings( Boss, data, return_defaults=True, ) elif statement_type == "insert_stmt": with self.sql_execution_asserter() as asserter: s.execute(insert(Boss), data) asserter.assert_( Conditional( testing.db.dialect.insert_executemany_returning, [ CompiledSQL( "INSERT INTO people (name, type) " "VALUES (:name, :type) RETURNING people.person_id", [ {"name": "b1", "type": "boss"}, {"name": "b2", "type": "boss"}, {"name": "b3", "type": "boss"}, ], ), ], [ CompiledSQL( "INSERT INTO people (name, type) " "VALUES (:name, :type)", [{"name": "b1", "type": "boss"}], ), CompiledSQL( "INSERT INTO people (name, type) " "VALUES (:name, :type)", [{"name": "b2", "type": "boss"}], ), CompiledSQL( "INSERT INTO people (name, type) " "VALUES (:name, :type)", [{"name": "b3", "type": "boss"}], ), ], ), CompiledSQL( "INSERT INTO managers (person_id, status, manager_name) " "VALUES (:person_id, :status, :manager_name)", [ {"person_id": 1, "status": "s1", "manager_name": "mn1"}, {"person_id": 2, "status": "s2", "manager_name": "mn2"}, {"person_id": 3, "status": "s3", "manager_name": "mn3"}, ], ), CompiledSQL( "INSERT INTO boss (boss_id, golf_swing) VALUES " "(:boss_id, :golf_swing)", [ {"golf_swing": "g1", "boss_id": 1}, {"golf_swing": "g2", "boss_id": 2}, {"golf_swing": "g3", "boss_id": 3}, ], ), ) @testing.combinations("update_mappings", "update_stmt") def test_bulk_update(self, statement_type): Person, Engineer, Manager, Boss = self.classes( "Person", "Engineer", "Manager", "Boss" ) s = fixture_session() b1, b2, b3 = ( Boss(name="b1", status="s1", manager_name="mn1", golf_swing="g1"), Boss(name="b2", status="s2", manager_name="mn2", golf_swing="g2"), Boss(name="b3", status="s3", manager_name="mn3", golf_swing="g3"), ) s.add_all([b1, b2, b3]) s.commit() # slight non-convenient thing. we have to fill in boss_id here # for update, this is not sent along automatically. this is not a # new behavior in bulk new_data = [ { "person_id": b1.person_id, "boss_id": b1.boss_id, "name": "b1_updated", "manager_name": "mn1_updated", }, { "person_id": b3.person_id, "boss_id": b3.boss_id, "manager_name": "mn2_updated", "golf_swing": "g1_updated", }, ] if statement_type == "update_mappings": with self.sql_execution_asserter() as asserter: s.bulk_update_mappings(Boss, new_data) elif statement_type == "update_stmt": with self.sql_execution_asserter() as asserter: s.execute(update(Boss), new_data) asserter.assert_( CompiledSQL( "UPDATE people SET name=:name WHERE " "people.person_id = :people_person_id", [{"name": "b1_updated", "people_person_id": 1}], ), CompiledSQL( "UPDATE managers SET manager_name=:manager_name WHERE " "managers.person_id = :managers_person_id", [ {"manager_name": "mn1_updated", "managers_person_id": 1}, {"manager_name": "mn2_updated", "managers_person_id": 3}, ], ), CompiledSQL( "UPDATE boss SET golf_swing=:golf_swing WHERE " "boss.boss_id = :boss_boss_id", [{"golf_swing": "g1_updated", "boss_boss_id": 3}], ), ) class BulkIssue6793Test(BulkTest, fixtures.DeclarativeMappedTest): __backend__ = True @classmethod def setup_classes(cls): Base = cls.DeclarativeBasic class User(Base): __tablename__ = "users" id = Column(Integer, Identity(), primary_key=True) name = Column(String(255), nullable=False) def test_issue_6793(self): User = self.classes.User session = fixture_session() with self.sql_execution_asserter() as asserter: session.bulk_save_objects([User(name="A"), User(name="B")]) session.add(User(name="C")) session.add(User(name="D")) session.flush() asserter.assert_( Conditional( testing.db.dialect.insert_executemany_returning, [ CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "A"}, {"name": "B"}], ), CompiledSQL( "INSERT INTO users (name) VALUES (:name) " "RETURNING users.id", [{"name": "C"}, {"name": "D"}], ), ], [ CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "A"}, {"name": "B"}], ), CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "C"}], ), CompiledSQL( "INSERT INTO users (name) VALUES (:name)", [{"name": "D"}], ), ], ) )