Skip to content

Instantly share code, notes, and snippets.

@daymien
Last active July 3, 2022 20:08
Show Gist options
  • Save daymien/6970291ab75048e6120775d1c1114a47 to your computer and use it in GitHub Desktop.
Save daymien/6970291ab75048e6120775d1c1114a47 to your computer and use it in GitHub Desktop.
SQLAlchemy ancestors, descendants query via recursive query with ORM Results
from sqlalchemy import Index, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
class Node(Base):
id = Column(Integer, primary_key=True, index=True)
parent_id = Column(ForeignKey("node.id"), nullable=True)
name = Column(String(256), index=False, nullable=False)
__table_args__ = (
Index("ix_node_parent", id, parent_id, unique=True),
)
def get_descendants(db: Session, node_id: int):
top_query = db.query(Node, literal(0).label("level")).filter(Node.id == node_id).cte("cte", recursive=True)
bottom_query = db.query(Node, (top_query.c.level + 1).label("level")).join(top_query, Node.parent_id == top_query.c.id)
recursive_query = top_query.union_all(bottom_query)
query = db.query(recursive_query).with_entities(Node).select_entity_from(recursive_query)
return query.all()
def get_ancestors(db: Session, node_id: int):
top_query = db.query(Node, literal(0).label("level")).filter(Node.id == node_id).cte("cte", recursive=True)
bottom_query = db.query(Node, (top_query.c.level + 1).label("level")).join(top_query, Node.id == top_query.c.parent_id)
recursive_query = top_query.union_all(bottom_query)
query = db.query(recursive_query).with_entities(Node).select_entity_from(recursive_query)
return query.all()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment