CODEWITHSUNDEEP
pythonpostgresqlsqlalchemy
yoyoy
yoyoy

Reputation: 395

SQLAlchemy Common Table Expression Returning Values

I would like to write a query like the following in SQLAlchemy:

WITH first_insert AS (
   INSERT INTO names (id, name)
   VALUES (1, 'bob')
   RETURNING id
)
INSERT ...

However, the following code gives me an error:

import sqlalchemy
from sqlalchemy import insert
from sqlalchemy.sql import table, column

print(sqlalchemy.__version__)

names = table("names", column("id"), column("name"))

stmt = insert(names).values(name="bob").returning(names.c.id)
print(stmt)

stmt = stmt.cte('names_cte')
print(stmt)

The output is

1.4.2
INSERT INTO names (name) VALUES (:name) RETURNING names.id

...

AttributeError: 'CTE' object has no attribute '_returning'

Is this possible in SQLAlchemy? I am working with PostgreSQL if that makes a difference.

Upvotes: 3

Views: 1220

Answers (1)

van
van

Reputation: 76972

You are right to find an error, but it seems to be limited only to the generation of the string version of the SQL at that particular point. To me it looks to be a bug, and it might make sense to create an Issue for it.

Nonetheless, if you can ignore that print statement, the rest of the code works quite well:

# added for dialect specific query printing
from sqlalchemy.dialects import postgresql
def print_stmt(stmt):
    print(stmt.compile(dialect=postgresql.dialect()))


# defined also second table "balls" for testing of INSERTs
names = table("names", column("id"), column("name"))
balls = table("balls", column("id"), column("name"))


# added multi-row insert to test    
stmt = insert(names).values([{"name": "bob"}, {"name": "alice"}]).returning(names.c.id, names.c.name)
# print(stmt)  # does not work with 'default' dialect for multi-row insert
print_stmt(stmt)


stmt = stmt.cte('names_cte')
# INDEED: putting together the string version of the statement generates an error,
# BUT: when further used, the error (which I think is a bug) is not in a way.
# print(stmt)


# insert from CET
inse = insert(balls).from_select(["id", "name"], stmt)
print_stmt(inse)

res = engine.execute(inse)
print(res.rowcount)


# test data is in the database    
q = session.query(balls)
for b in q:
    print(b)

Upvotes: 4

Related Questions