Linked list recursive query

[khoroshevskyi]

Hi!
I am trying to build a query that will recursively return all rows in the table that are connected by self referential.
The table looks like that:

class Samples(Base):
    """
    Samples table representation in the database
    """

    __tablename__ = "samples"

    guid: Mapped[int] = mapped_column(primary_key=True)
    sample: Mapped[dict] = mapped_column(JSON, server_default=FetchedValue())
    project_id = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"))
    project_mapping: Mapped["Projects"] = relationship(back_populates="samples_mapping")
    sample_name: Mapped[Optional[str]] = mapped_column()

    parent_guid: Mapped[Optional[int]] = mapped_column(ForeignKey("samples.guid", ondelete="CASCADE"),
                                                        nullable=True,
                                                        doc="Parent sample id. Used to create a hierarchy of samples.")

    parent_mapping: Mapped["Samples"] = relationship("Samples", remote_side=guid, back_populates="child_mapping")
    child_mapping: Mapped["Samples"] = relationship("Samples", back_populates="parent_mapping")

    views: Mapped[Optional[List["ViewSampleAssociation"]]] = relationship(
        back_populates="sample", cascade="all, delete-orphan"
    )

parent_guid points to his parent id. All parents can have only one child and one child can have only one parent. There will be always root sample with parent_guid = None.

I am trying to recursively get all rows that are connected. I can do it in sql, but can't figure out how I can do it in sqlalchemy.
This is sql query:

WITH RECURSIVE sample_hierarchy AS (
    -- Anchor member: Select the root sample(s)
    SELECT
        s.guid,
        s.sample,
        s.row_number,
        s.project_id,
        s.sample_name,
        s.parent_guid,
        0 AS depth
    FROM
        samples s
    WHERE
        s.project_id = 2
        AND s.parent_guid IS NULL  -- Root samples

    UNION ALL

    -- Recursive member: Select child samples
    SELECT
        s.guid,
        s.sample,
        s.row_number,
        s.project_id,
        s.sample_name,
        s.parent_guid,
        sh.depth + 1 AS depth
    FROM
        samples s
    JOIN
        sample_hierarchy sh ON s.parent_guid = sh.guid
)
SELECT *
FROM sample_hierarchy
    order by depth;

Do you have any suggestions how to build such a query in sqlalchemy and what is the most efficient way?
Thank you for your help!

[gordthompson]

Perhaps this simplified example might give you an idea on how to proceed:

from pprint import pprint
from typing import Optional

from sqlalchemy import create_engine, ForeignKey, insert, select, literal_column
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session


class Base(DeclarativeBase):
    pass


class Person(Base):
    __tablename__ = "person"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=False)
    name: Mapped[str]
    parent_id: Mapped[Optional[int]] = mapped_column(ForeignKey("person.id"))


engine = create_engine("postgresql://scott:tiger@192.168.0.199/test")
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

person_table = Person.__table__

# populate example data
with engine.begin() as conn:
    conn.execute(
        insert(person_table),
        [
            dict(id=1, name='Abe "Grampa" Simpson', parent_id=None),
            dict(id=2, name="Homer Simpson", parent_id=1),
            dict(id=3, name="Bart Simpson", parent_id=2),
        ],
    )

my_cte = (
    select(
        person_table.c.id,
        person_table.c.name,
        person_table.c.parent_id,
        literal_column("0").label("depth"),
    )
    .where(person_table.c.parent_id.is_(None))
    .cte(name="my_cte", recursive=True)
)
my_cte_alias = my_cte.alias(name="my_cte_alias")
my_cte = my_cte.union_all(
    select(
        person_table.c.id,
        person_table.c.name,
        person_table.c.parent_id,
        (my_cte_alias.c.depth + 1).label("depth"),
    ).join(my_cte_alias, person_table.c.parent_id == my_cte_alias.c.id)
)
stmt = select(my_cte.c.name, my_cte.c.depth).order_by(my_cte.c.depth)
with Session(engine) as sess:
    rows = sess.execute(stmt).mappings().all()
    pprint(rows, width=20)
    """
[{'name': 'Abe "Grampa" Simpson', 'depth': 0},
 {'name': 'Homer Simpson', 'depth': 1},
 {'name': 'Bart Simpson', 'depth': 2}]
    """

[khoroshevskyi]

Thank you, it seems like a good solution!