[sqlalchemy] common table expressions for INSERT, UPDATE, and DELETE

[Hunter Blanks]

Hi,

Although SQLAlchemy supports PostgreSQL's common table expressions (i.e. 
WITH statements*)  for SELECT, it does not yet seem to support them for 
INSERT, UPDATE, or DELETE.

(1) Can anyone confirm that this is the case? Lest it help, I've attached 
test cases for UPDATE and DELETE below.

(2) If this is the case, can anyone offer a guess for what it would involve 
for me to add these features?

I know that in most cases, people are able to work around this by passing 
around aliased select() clauses – but it sure would be nice to have the use 
of CTE's, too. When viewing the SQL, they do make for better legibility, 
and they can also help with query optimization.

Thanks much,

Hunter Blanks

* Syntax can be found in PostgreSQL's documentation for 
INSERT<http://www.postgresql.org/docs/9.3/static/sql-insert.html>
, UPDATE <http://www.postgresql.org/docs/9.3/static/sql-update.html>, and 
DELETE <http://www.postgresql.org/docs/9.3/static/sql-delete.html>.


"""
(Failing) test cases for UPDATE and DELETE statements with PostgreSQL's
common table expressions.
"""

import sqlalchemy
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import select
from sqlalchemy.sql.expression import exists


SRC_TABLE_ROWS = [
    {'id': i, 'name': name}
    for i, name in enumerate(
        ['zero', 'one', 'two', 'three', 'four', 'five']
    )
]

DST_TABLE_ROWS = [{'id': i, 'name': 'x'} for i in range(6)]


def create_tmp_tables(connection):
    """ Creates temporary tables for the test case. """
    metadata = sqlalchemy.MetaData()

    tmp_src_table = sqlalchemy.Table(
        'tmp_src_table', metadata,
        Column('id', Integer, primary_key=True),
        Column('name', String(255), nullable=False),
        prefixes=['TEMPORARY']
    )

    tmp_dst_table = sqlalchemy.Table(
        'tmp_dst_table', metadata,
        Column('id', Integer, primary_key=True),
        Column('name', String(255), nullable=False),
        prefixes=['TEMPORARY']
    )

    metadata.create_all(connection)

    connection.execute(tmp_src_table.insert(), SRC_TABLE_ROWS)
    connection.execute(tmp_dst_table.insert(), DST_TABLE_ROWS)

    return tmp_src_table, tmp_dst_table



def test_cte_update(connection):
    """
    Tests an UPDATE statement using a CTE.
    
    Args:
    
        - connection: SQLAlchemy connection with permissions to create
          temporary tables
    """
    src_table, dst_table = create_tmp_tables(connection)
    
    even_rows = src_table.select().where(
        src_table.c.id % 2 == 0
    ).cte('even_rows')

    # Verify working CTE    
    result = connection.execute(even_rows.select())

    update = dst_table.update().values(
        name=even_rows.c.name
    ).where(
        dst_table.c.id == even_rows.c.id
    )

    connection.execute(update)


def test_cte_delete(connection):
    """
    Tests an DELETE statement using a CTE.
    
    Args:
    
        - connection: SQLAlchemy connection with permissions to create
          temporary tables
    """
    src_table, dst_table = create_tmp_tables(connection)
    
    even_rows = src_table.select().where(
        src_table.c.id % 2 == 0
    ).cte('even_rows')

    # Verify working CTE    
    result = connection.execute(even_rows.select())

    delete_even_rows = dst_table.delete().where(
        exists().where(
            dst_table.c.id == even_rows.c.id
        )
    )

    connection.execute(delete_even_rows)

-- 
You received this message because you are subscribed to the Google Groups 
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.