Method test_upsert_from_select

test/sql/test_cte.py:1281–1332

(self)

        )

    def test_upsert_from_select(self):
        orders = table(
            "orders",
            column("region"),
            column("amount"),
            column("product"),
            column("quantity"),
        )

        upsert = (
            orders.update()
            .where(orders.c.region == "Region1")
            .values(amount=1.0, product="Product1", quantity=1)
            .returning(*(orders.c._all_columns))
            .cte("upsert")
        )

        insert = orders.insert().from_select(
            orders.c.keys(),
            select(
                literal("Region1"),
                literal(1.0),
                literal("Product1"),
                literal(1),
            ).where(~exists(upsert.select())),
        )

        self.assert_compile(
            insert,
            "WITH upsert AS (UPDATE orders SET amount=:param_5, "
            "product=:param_6, quantity=:param_7 "
            "WHERE orders.region = :region_1 "
            "RETURNING orders.region, orders.amount, "
            "orders.product, orders.quantity) "
            "INSERT INTO orders (region, amount, product, quantity) "
            "SELECT :param_1 AS anon_1, :param_2 AS anon_2, "
            ":param_3 AS anon_3, :param_4 AS anon_4 WHERE NOT (EXISTS "
            "(SELECT upsert.region, upsert.amount, upsert.product, "
            "upsert.quantity FROM upsert))",
            checkparams={
                "param_1": "Region1",
                "param_2": 1.0,
                "param_3": "Product1",
                "param_4": 1,
                "param_5": 1.0,
                "param_6": "Product1",
                "param_7": 1,
                "region_1": "Region1",
            },
        )

        eq_(insert.compile().isinsert, True)

    @testing.combinations(
        ("default_enhanced",),