How do I use the SQLAlchemy ORM to do an insert with a subquery (moving data from one table to another)

Question

The following simple example describes my problem with my postgres DB (although my question is more about sqlalchemy than postgres):

I have a table called detection with columns:

• id
• item
• price_in_cents
• shop_id

I have another table called item with the following columns:

• id
• detection_id (foreign key to detection.id)
• price_in_dollar

I want to move the entire dataset of a certain shop from table detection to table item whilst also performing an operation to convert cents to dollars (the example is theoretical, my real problem has a different operation than cents to dollars).

In raw SQL I can use the following query:

INSERT INTO item (detection_id, price_in_dollar) 
    SELECT id AS detection_id, 
    price_in_cent / 100 AS price_in_dollar 
    FROM detection 
    WHERE shop_id = {shop_id}

Is it possible to replicated this query using SQLAlchemy? Due to the volume of the data (could be millions of rows) I do not want to first download the data to do the operation and then upload it. My example that would work would be:

q = session.query(Detection).filter(Detection.shop_id == shop_id)
for detection_record in q:
    session.add(Item(detection_id=detection_record.id,
                     price_in_dollar=detection_record.price_in_cent / 100))
session.commit()

This would however download all the data to the machine first instead of doing all the work in the DB itself and thus has different behaviour than my example query.

Answer 1

Just because you're using ORM in your project doesn't mean that you have to use ORM for everything. SQLAlchemy ORM is good for pulling relational "things" down as Python objects and working with them. For server-side operations, SQLAlchemy Core is the tool to use.

Assuming that you have declared your ORM objects using Declarative, e.g.,

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class Detection(Base):
    __tablename__ = "detection"
    # ...

then you can use Core to create the server-side operation with code like this:

meta = Base.metadata
item_t = meta.tables[Item.__tablename__]
detection_t = meta.tables[Detection.__tablename__]

target_shop_id = 1  # test value

ins = item_t.insert().from_select(
    ["detection_id", "price_in_dollar"],
    sa.select(
        [
            detection_t.c.id.label("detection_id"),
            (detection_t.c.price_in_cents / sa.text("100")).label(
                "price_in_dollar"
            ),
        ]
    ).where(detection_t.c.shop_id == target_shop_id),
)

with engine.begin() as conn:
    conn.execute(ins)

and the generated SQL text is

INSERT INTO item (detection_id, price_in_dollar) SELECT detection.id AS detection_id, detection.price_in_cents / 100 AS price_in_dollar 
FROM detection 
WHERE detection.shop_id = ?