[Python] A recursive property defined as a property_column in SQLAlchemy

I am looking for a proper way to manage a recursive object using SQLAlchemy.

The column_property syntax looks very promising, but I could not make it work to load the children.

Below is an MRE:

from sqlalchemy import (
Column,
ForeignKey,
Integer,
create_engine,
select,
)
from sqlalchemy.orm import (
backref,
column_property,
declarative_base,
relationship,
remote,
Session,
)
from sqlalchemy.sql.expression import null
from testcontainers.postgres import PostgresContainer


Base = declarative_base()


class Category(Base):
__tablename__ = "categories"

id = Column(Integer, primary_key=True)
parent_id = Column(Integer, ForeignKey("categories.id"), index=True, nullable=True)

children = relationship(
"Category",
primaryjoin=(id == remote(parent_id)),
lazy="select",
backref=backref(
"parent",
primaryjoin=(remote(id) == parent_id),
lazy="select",
),
)

deep_children = column_property(null())


def build_deep_children_expr():
cte_query = select(Category.id).cte(recursive=True, name="descendants_cte")

parent_query = select(Category).where(Category.parent_id == cte_query.c.id)

cte_query = cte_query.union_all(parent_query)

return select(cte_query).scalar_subquery()


Category.deep_children = column_property(build_deep_children_expr())

if __name__ == "__main__":
# Use a temporary PostgreSQL container
with PostgresContainer("postgres:16.1") as postgres:
database_url = postgres.get_connection_url(driver="psycopg")
engine = create_engine(database_url, echo=True)

Base.metadata.create_all(engine)

session = Session(engine)

# Build a hierarchy of categories:
# root (0)
# ├─ (1)
# └─ (2)
# ├─ (3)
# ├─ (4)
# └─ (5)
# ├─ (6)
# └─ (7)
root = Category(
id=0,
children=[
Category(id=1),
Category(
id=2,
children=[
Category(id=3),
Category(id=4),
Category(id=5, children=[Category(id=6), Category(id=7)]),
],
),
],
)

session.add(root)
session.flush()
session.expire_all()

cat = session.query(Category).filter(Category.id == 0).one()

print(f"Category: {cat!r}")
print(f" deep_children_count: {cat.deep_children_count}")
print(f" children: {[c.id for c in cat.children]}")

# Check each child's deep_children too
for c in cat.children:
print(f" Child {c.id} => deep_children = {c.deep_children}")

session.close()


Obtained error:

sqlalchemy.exc.CompileError: All selectables passed to CompoundSelect must have identical numbers of columns; select #1 has 1 columns, select #2 has 3

Continue reading...