Postgresql - sqlmodel query with a subselect self join

user1931485
April 25, 2024
131 views
0 votes
2 Answers

Trying to do a self join on a model but am getting confused how to do so in sqlmodel ORM. Something equivalent to the use of following SQL in Postgresql.

SELECT tt.id, ttp.parent_name, tt.name
  FROM target_table tt, 
       (select DISTINCT bb.parent_id, bbref.name parent_name
          from target_table tt, target_table ttref
         WHERE tt.parent_id is not null 
           AND tt.parent_id = ttref.id
         order by 1) ttp
 WHERE tt.parent_id = ttp.parent_id
ORDER BY 1

I have a corresponding object built using sqlmodel ORM, like:

class TargetTable(SQLModel, table=True):...

Works fine for normal read / write.

with Session(engine) as session:
    print(select(TargetTable))
    for tt in session.exec(select(TargetTable)):
        print(tt)

I now need to do an operation for which I need to retrieve some data as described in the SQL above. I am not sure – rather I do not know how to construct the definition and corresponding join with the second table.

I was trying along the lines of:

select(TargetTable).join(select(TargetTable.label('tt')).join(
        (TargetTable.label('ttref').filter(tt.parent_id != NULL, tt.parent_id == ttref.id)
        ).label(ttp)).filter(tt.parent_id = ttp.parent_id)

But it is not working and I am getting confused in exactly how to go about it.

Appreciate any pointers or a solution.

Answers

Chosen as BEST ANSWER

*** UPDATE: Adding the full non-working code in response to @python_user ask. I tried to follow the response advice from python_user to just include the query in the code here.

### 2. ONLY USING SQLMODEL but with the query as advice by @python_user on SO - NOT WORKING!!
# below code block for hierarchy modeling was my attempt to use all sqlmodel class and properties
# did not go well
# ERROR:
#     Traceback(most recent call last):
#         File "/Users/tapas/PycharmProjects/ResiChainLab/test/checkmodel.py", line 38, in < module >
#             ttp = alias((select(tt.parent_id, ttref.name.label('parent_name')).distinct().join(ttref, (tt.parent_id is not None)
#                                 ^^^^^^^^^^^^
#     AttributeError: 'Alias' object has no attribute 'parent_id'
###

from typing import Optional

from sqlmodel import SQLModel, create_engine, Session, select
from sqlmodel import Field, alias

engine = create_engine("sqlite:///cptest.db", echo=True)


class TargetTable(SQLModel, table=True):
    __tablename__ = "target_table"
    id: Optional[int] = Field(primary_key=True)
    parent_id: int | None = Field(foreign_key='target_table.id')
    name: str


SQLModel.metadata.create_all(bind=engine)

with Session(engine) as session:
    session.add(TargetTable(name="name 1"))
    session.add(TargetTable(name="name 2", parent_id=1))
    session.add(TargetTable(name="name 3", parent_id=2))
    session.commit()

with Session(engine) as session:
    tt = alias(TargetTable, 'tt')  # sqlmodel alias()
    ttref = alias(TargetTable, 'ttref')
    ttp = alias((select(tt.parent_id, ttref.name.label('parent_name'))
                 .distinct()
                 .join(ttref, (tt.parent_id is not None) & (tt.parent_id == ttref.id))
                 .subquery()), 'ttp', True
                )
    resultset = session.execute(select(tt.id, ttp.c.parent_name, tt.name)
                                .join(ttp, tt.parent_id == ttp.c.parent_id)
                                .order_by(tt.id)
                                )
    for result in resultset:
        print(result)

So, I changed the above to only include sqlalchemy for the aliasing and specifying the order_by indices from @python_users answer, and now it works for most part - except that for some reasons, it seems to be ignoring for the not null filter on the join.

The latest code is:

#### FINAL - HYBRID & WORKING
# uses sqlalchemy for aliases and specifying order by indices with text()

from typing import Optional

from sqlmodel import SQLModel, create_engine, Session, select
from sqlmodel import Field

from sqlalchemy import text
from sqlalchemy.orm import aliased

engine = create_engine("sqlite:///cptest-2.db", echo=True)


class TargetTable(SQLModel, table=True):
    __tablename__ = "target_table"
    id: Optional[int] = Field(primary_key=True)
    parent_id: int | None = Field(foreign_key='target_table.id')
    name: str


SQLModel.metadata.create_all(bind=engine)

with Session(engine) as session:
    session.add(TargetTable(name="name 1"))
    session.add(TargetTable(name="name 2", parent_id=1))
    session.add(TargetTable(name="name 3", parent_id=2))
    session.commit()

with Session(engine) as session:
    tt = aliased(TargetTable, name="tt")
    ttref = aliased(TargetTable, name="ttref")
    ttp = (select(tt.parent_id, ttref.name.label("parent_name"))
           .distinct()
           .filter(tt.parent_id == ttref.id, tt.parent_id is not None)
           .order_by(text("1"))
           .subquery()
           .alias("ttp")
           )
    resultset = session.execute(select(tt.id, ttp.c.parent_name, tt.name)
                                .join(ttp, tt.parent_id == ttp.c.parent_id)
                                .order_by(text("1"))
                                )
    for result in resultset:
        print(result)

(Edit)

There is nothing SQLModel specific in this question, you have to use SQLAlchemy to do this. This approach emits the following query.

SELECT 
  tt.id, 
  ttp.parent_name, 
  tt.name 
FROM 
  target_table AS tt 
  JOIN (
    SELECT 
      DISTINCT tt.parent_id AS parent_id, 
      ttref.name AS parent_name 
    FROM 
      target_table AS tt 
      JOIN target_table AS ttref ON tt.parent_id IS NOT NULL 
      AND tt.parent_id = ttref.id 
    ORDER BY 
      1
  ) AS ttp ON tt.parent_id = ttp.parent_id 
ORDER BY 
  1

Here is a complete code that has the model defined in SQLAlchemy, but you should just be able to copy paste the query part and use with SQLModel.

from sqlalchemy import ForeignKey, create_engine, null, select, text
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, aliased, mapped_column

class Base(DeclarativeBase):
    pass

engine = create_engine("sqlite:///temp.db", echo=True)

class TargetTable(Base):
    __tablename__ = "target_table"
    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int | None] = mapped_column(ForeignKey(id))
    name: Mapped[str]

Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(TargetTable(name="name 1"))
    session.add(TargetTable(name="name 2", parent_id=1))
    session.add(TargetTable(name="name 3", parent_id=2))
    session.commit()

with Session(engine) as session:
    tt = aliased(TargetTable, name="tt")
    ttref = aliased(TargetTable, name="ttref")
    ttp = (
        select(tt.parent_id, ttref.name.label("parent_name"))
        .distinct()
        .join(ttref, (tt.parent_id != null()) & (tt.parent_id == ttref.id))
        .order_by(text("1"))
        .subquery()
        .alias("ttp")
    )
    statement = (
        select(tt.id, ttp.c.parent_name, tt.name)
        .join(ttp, tt.parent_id == ttp.c.parent_id)
        .order_by(text("1"))
    )
    for i in session.execute(statement):
        print(i)

Please signup or login to give your own answer.

Click here to cancel reply.

Postgresql – sqlmodel query with a subselect self join

Answers