Method test_bulk_update

test/orm/test_composites.py:454–520

(self, type_)

    @testing.combinations("legacy", "statement")
    def test_bulk_update(self, type_):
        Edge, Point = (self.classes.Edge, self.classes.Point)
        Graph = self.classes.Graph

        sess = self._fixture()

        graph = Graph(id=2)
        sess.add(graph)
        sess.flush()
        graph_id = 2

        data = [
            {
                "start": Point(random.randint(1, 50), random.randint(1, 50)),
                "end": Point(random.randint(1, 50), random.randint(1, 50)),
                "graph_id": graph_id,
            }
            for i in range(25)
        ]
        sess.execute(insert(Edge), data)

        inserted_data = [
            dict(row._mapping)
            for row in sess.execute(
                select(Edge.id, Edge.start, Edge.end, Edge.graph_id)
                .where(Edge.graph_id == graph_id)
                .order_by(Edge.id)
            )
        ]

        to_update = []
        updated_pks = {}
        for rec in random.choices(inserted_data, k=7):
            rec_copy = dict(rec)
            updated_pks[rec_copy["id"]] = rec_copy
            rec_copy["start"] = Point(
                random.randint(1, 50), random.randint(1, 50)
            )
            rec_copy["end"] = Point(
                random.randint(1, 50), random.randint(1, 50)
            )
            to_update.append(rec_copy)

        expected_dataset = [
            updated_pks[row["id"]] if row["id"] in updated_pks else row
            for row in inserted_data
        ]

        if type_ == "statement":
            sess.execute(update(Edge), to_update)
        elif type_ == "legacy":
            sess.bulk_update_mappings(Edge, to_update)
        else:
            assert False

        edges = self.tables.edges
        eq_(
            sess.execute(