带有子选择自连接的sqlmodel查询

问题描述 投票：0回答：2

尝试在模型上进行自连接，但我很困惑如何在 sqlmodel ORM 中执行此操作。相当于在 Postgresql 中使用以下 SQL 的东西。

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

我有一个使用 sqlmodel ORM 构建的相应对象，例如：

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

正常读/写工作正常。

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

我现在需要执行一项操作，需要检索上面 SQL 中所述的一些数据。我不确定 - 相反，我不知道如何构建定义以及与第二个表的相应连接。

我正在尝试：

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)

但它不起作用，我对到底如何去做感到困惑。

感谢任何指点或解决方案。

python python-3.x postgresql sqlalchemy sqlmodel
2个回答
0
投票

这个问题没有任何特定于 SQLModel 的内容，您必须使用 SQLAlchemy 来执行此操作。这种方法会发出以下查询。

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

这是一个完整的代码，其中在 SQLAlchemy 中定义了模型，但您应该能够复制粘贴查询部分并与 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)
0
投票

*** 更新：添加完整的非工作代码以响应@python_user 询问。我尝试遵循 python_user 的响应建议，将查询包含在此处的代码中。

### 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)

因此，我将上面的内容更改为仅包含 sqlalchemy 用于别名并指定 @python_users 答案中的 order_by 索引，现在它在大多数情况下都有效 - 除了由于某些原因，它似乎忽略了非空过滤器加入。

最新代码是：

#### 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)
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.