Issue: Support CTE in ModelInsert?
Author: albireox
State: closed
Created: 2020-04-13 23:54:42

I have the following use case. I have a table `tic_v8` from which I want to get the columns `ra` and `dec` and insert them into a table `catalog`, also assigning a serial primary key (`catalogid`). In the same query I want to retrieve the newly assigned catalogids and `tic_v8.id` and add them to a relational table `catalog_to_tic_v8`. My real query is a bit more complicated but this should illustrate.

The only way I've found to do this (I'd be interested to know if there is a better way) is to first populate the relational table with a sequential `catalogid` and then populate `catalog`. I can do this in SQL with a couple CTEs

```sql
WITH x AS (SELECT * FROM tic_v8 ORDER BY id ASC), 
	y AS (INSERT INTO catalog_to_tic_v8 (catalogid, target_id) SELECT ROW_NUMBER() OVER(), id FROM x ORDER BY x.id RETURNING catalogid, target_id) 
INSERT INTO catalog (catalogid, ra, dec) SELECT catalogid, t.ra, t.dec FROM y join tic_v8 t ON y.target_id = t.id;
```

(I understand that for this one-to-one relationship I would not need a relational table, but in a different query I'll need to populate this table with one-to-many cases).

In Peewee, if I try to do something like this

```python
x = TIC.select().cte('x')
y = CatalogToTIC_v8.insert_from(x.select(fn.row_number().over(), x.c.id).order_by(x.c.id.asc()), [CatalogToTIC_v8.catalogid, CatalogToTIC_v8.target_id]).returning(CatalogToTIC_v8)).cte('y')
q = Catalog.insert_from(y.select(y.c.catalogid, TIC.ra, TIC.dec).join(TIC, on=y.c.target_id==TIC.id), [Catalog.catalogid, Catalog.ra, Catalog.dec]).returning().with_cte(x, y)
```

that would fail because a `ModelInsert` cannot be a CTE. PostgreSQL allows it and actually I can do

```python
y = peewee.CTE('y', CatalogToTIC_v8.insert_from(x.select(fn.row_number().over(), x.c.id).order_by(x.c.id.asc()), [CatalogToTIC_v8.catalogid, CatalogToTIC_v8.target_id]).returning(CatalogToTIC_v8)))
q = Catalog.insert_from(y.select(y.c.catalogid, TIC.ra, TIC.dec).join(TIC, on=y.c.target_id==TIC.id), [Catalog.catalogid, Catalog.ra, Catalog.dec]).returning().with_cte(x, y)
```

Would it make sense to add the `.cte()` method to the `ModelInsert`?

This is now implemented in 21dcc113c28b853544ca1db804750320d934e17e

Examples can be found in the new tests, reproduced here:

```python
    def test_data_modifying_cte_delete(self):
        # delete products whose price is less than 3, simultaneously copying them to
        # the archive table.
        query = (C_Product.delete()
                 .where(C_Product.price < 3)
                 .returning(C_Product.id, C_Product.name, C_Product.price))
        cte = query.cte('moved_rows')

        src = Select((cte,), (cte.c.id, cte.c.name, cte.c.price))
        res = (C_Archive
               .insert_from(src, (C_Archive.id, C_Archive.name, C_Archive.price))
               .with_cte(cte)
               .execute())
        self.assertEqual(len(list(res)), 3)  # The 3 products were moved successfully.

        self.assertEqual(
            sorted([(p.name, p.price) for p in C_Product.select()]),
            [('p3', 3), ('p4', 4)])
        self.assertEqual(
            sorted([(p.name, p.price) for p in C_Archive.select()]),
            [('p0', 0), ('p1', 1), ('p2', 2)])

        # Use a recursive CTE to bulk-delete a set of parts related by a heirarchical
        # foreign-key.
        base = (C_Part
                .select(C_Part.sub_part, C_Part.part)
                .where(C_Part.part == 'mp1')
                .cte('included_parts', recursive=True,
                     columns=('sub_part', 'part')))
        PA = C_Part.alias('p')
        recursive = (PA
                     .select(PA.sub_part, PA.part)
                     .join(base, on=(PA.part == base.c.sub_part)))
        cte = base.union_all(recursive)

        sq = Select((cte,), (cte.c.part,))
        res = (C_Part.delete()
               .where(C_Part.part.in_(sq))
               .with_cte(cte)
               .execute())

        self.assertEqual(sorted([p.part for p in C_Part.select()]),
                         ['mp2', 'mp2-c', 'mp2-c-g'])

    def test_data_modifying_cte_update(self):
        # Populate archive table w/copy of data in product.
        C_Archive.insert_from(
            C_Product.select(),
            (C_Product.id, C_Product.name, C_Product.price)).execute()

        # Double the price on all products in a CTE and select updated results.
        query = (C_Product
                 .update(price=C_Product.price * 2)
                 .returning(C_Product.id, C_Product.name, C_Product.price))
        cte = query.cte('t')

        sq = cte.select_from(cte.c.id, cte.c.name, cte.c.price)
        self.assertEqual(sorted([(x.name, x.price) for x in sq]), [
            ('p0', 0), ('p1', 2), ('p2', 4), ('p3', 6), ('p4', 8)])

        # Ensure changes were persisted.
        self.assertEqual(sorted([(x.name, x.price) for x in C_Product]), [
            ('p0', 0), ('p1', 2), ('p2', 4), ('p3', 6), ('p4', 8)])

        # Double the prices again in a CTE, and simultaneously update
        # all the prices in the archive table.
        sq = Select((cte,), (cte.c.id, cte.c.price))
        res = (C_Archive
               .update(price=sq.c.price)
               .from_(sq)
               .where(C_Archive.id == sq.c.id)
               .with_cte(cte)
               .execute())

        self.assertEqual(sorted([(x.name, x.price) for x in C_Product]), [
            ('p0', 0), ('p1', 4), ('p2', 8), ('p3', 12), ('p4', 16)])
        self.assertEqual(sorted([(x.name, x.price) for x in C_Archive]), [
            ('p0', 0), ('p1', 4), ('p2', 8), ('p3', 12), ('p4', 16)])

    def test_data_modifying_cte_insert(self):
        # Insert a new row within a CTE and select the resulting new row data.
        query = (C_Product
                 .insert({'name': 'p5', 'price': 5})
                 .returning(C_Product.id, C_Product.name, C_Product.price))
        cte = query.cte('t')

        sq = cte.select_from(cte.c.id, cte.c.name, cte.c.price)
        self.assertEqual([(p.name, p.price) for p in sq], [('p5', 5)])

        # Insert a new row within a CTE and simultaneously insert all rows 
        # from the product table into the archive.
        query = (C_Product
                 .insert({'name': 'p6', 'price': 6})
                 .returning(C_Product.id, C_Product.name, C_Product.price))
        cte = query.cte('t')
        sq = Select((cte,), (cte.c.id, cte.c.name, cte.c.price))
        res = (C_Archive
               .insert_from(sq, (sq.c.id, sq.c.name, sq.c.price))
               .with_cte(cte)
               .execute())
        self.assertEqual([(p.name, p.price) for p in C_Archive], [('p6', 6)])

        self.assertEqual(sorted([(p.name, p.price) for p in C_Product]), [
            ('p0', 0), ('p1', 1), ('p2', 2), ('p3', 3), ('p4', 4), ('p5', 5),
            ('p6', 6)])
```

@coleifer 

Way to go! This one is great.

Hi @coleifer 

How can we populate a model with a select from a CTE?

To take a trivial example:

```
user_select = User.select().where(
    User.email == email
).cte('user_select')
```

How can this be mapped back onto a User model?

I would like the query to resolve to the following, while the ORM maps it back to User:

```
WITH user_select AS (select * from user where email = 'foo')
SELECT * from user_select
```

This kind of question typically belongs on stackoverflow, as it is covered by the docs already (and is not even related to this github issue!). Docs: http://docs.peewee-orm.com/en/latest/peewee/querying.html#common-table-expressions

To your example:

```python

class User(Model):
    name = TextField()
    class Meta:
        database = db

cte = User.select().where(User.name == 'c').cte('user_select')
sq = User.select(cte.c.name).from_(cte).with_cte(cte)
for row in sq:
    print(row.name)
```