Last active
July 3, 2022 20:08
-
-
Save daymien/6970291ab75048e6120775d1c1114a47 to your computer and use it in GitHub Desktop.
SQLAlchemy ancestors, descendants query via recursive query with ORM Results
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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