Method test_bulk_insert

test/orm/test_composites.py:326–375

(self, type_)

        ("values_returning", testing.requires.insert_returning),
    )
    def test_bulk_insert(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)
        ]
        returning = False
        if type_ == "statement":
            sess.execute(insert(Edge), data)
        elif type_ == "stmt_returning":
            result = sess.scalars(insert(Edge).returning(Edge), data)
            returning = True
        elif type_ == "values":
            sess.execute(insert(Edge).values(data))
        elif type_ == "values_returning":
            result = sess.scalars(insert(Edge).values(data).returning(Edge))
            returning = True
        elif type_ == "legacy":
            sess.bulk_insert_mappings(Edge, data)
        else:
            assert False

        if returning:
            eq_(result.all(), [Edge(rec["start"], rec["end"]) for rec in data])

        edges = self.tables.edges
        eq_(
            sess.execute(
                select(edges.c["x1", "y1", "x2", "y2"])
                .where(edges.c.graph_id == graph_id)
                .order_by(edges.c.id)
            ).all(),
            [
                (e["start"].x, e["start"].y, e["end"].x, e["end"].y)
                for e in data
            ],
        )

    @testing.combinations("legacy", "statement")
    def test_bulk_insert_heterogeneous(self, type_):