如何使用 SQLAlchemy 在特定级别建立节点和后代之间的关系

Question

我有一个包含分层数据的表，其中每一行包含一个 id、parent_id、level（深度）以及一个路径枚举 hierarchy_hint，它是一个具有类似 JSON 格式的纯字符串。例如：

id	parent_id	level	hierarchy_hint
1	null	'level1'	'{"level1": "root"}'
2	1	'level2'	'{"level1": "root", "level2: "shoes"}'
3	2	'level3'	'{"level1": "root", "level2: "shoes", "level3": "athletic"}'

（例如，我使用整数，但 id 是 UUID）

我想使用hierarchy_hint 在特定level 处获取给定Node 的所有后代。该模型已经建立了自引用关系以获取直系后代 (children)。模型如下：

class Node(db.Model):
    id = db.Column(UUID(as_uuid=True), primary_key=True)
    parent_id = db.Column(UUID(as_uuid=True), db.ForeignKey('node.id'))
    children = db.relationship('Node')
    level = db.Column(db.String(length=4000), primary_key=False)
    hierarchy_hint = db.Column(db.String(length=200), primary_key=False)

我已经尝试添加到模型中：

    level_threes = relationship(
        "Node", primaryjoin=and_(
            foreign(hierarchy_hint).like(func.regexp_replace(hierarchy_hint, '}', ',%%')),
            level == 'level3'), viewonly=True)

但我得到了错误：

sqlalchemy.exc.ArgumentError: Relationship Node.level_threes could not determine any unambiguous local/remote column pairs based on join condition and remote_side arguments.  Consider using the remote() annotation to accurately mark those elements of the join condition that are on the remote side of the relationship.

我怀疑因为我正在使用regexp_replace 修改hierarchy_hint，所以我应该以某种方式使用secondary 表，但我对SQLAlchemy 的经​​验不足，不知道该怎么做。

我也愿意接受有关如何使用递归查询或其他方式执行此操作的替代建议，但我希望能够以与访问 children 相同的方式访问该属性，以避免在其他地方重构。

我正在使用 Flask、SQLAlchemy 和 PostgreSQL。

在纯 SQL 中，我尝试做的可能如下所示：

with ancestor as (select hierarchy_hint from node where id='2')
select node.id from node, ancestor
    where node.hierarchy_hint and node.level='level4';

在这里，我们希望得到shoes 树中的所有level4 项目，但没有来自任何其他树的level4 项目，也没有来自shoes 的任何level5 等项目。

Answer 1

我实际上认为我知道你想要做什么。我认为您可以执行以下操作。您可能知道，但您可能需要仔细检查您在hierarchy_hint 中的路径是否具有确定性，因为我不确定常规 JSON 是否始终以相同的顺序放置密钥。不过，当我测试它时，它似乎与转储一起工作。我尝试添加一些数据以确保它不会返回同一级别的所有内容。

别名关系

class Node(Base):
    __tablename__ = 'nodes'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('nodes.id'))
    parent = relationship('Node', backref='children', remote_side='nodes.c.id')
    level = Column(String(length=200), primary_key=False)
    name = Column(String(length=200), primary_key=False)
    hierarchy_hint = Column(String(length=4000), primary_key=False)

# After class is constructed.
aliased_node = aliased(Node)

# Attach the relationship after class is constructed so we can use alias.
Node.level_threes = relationship(
    aliased_node, primaryjoin=and_(
        aliased_node.hierarchy_hint.like(func.regexp_replace(Node.hierarchy_hint, '}', ',%%')),
        aliased_node.level == 'level3'),
    viewonly=True,
    uselist=True,
    remote_side=[aliased_node.level, aliased_node.hierarchy_hint],
    foreign_keys=Node.hierarchy_hint)

Base.metadata.create_all(engine)

node_values = [
    ('level1', 'root', {"level1": "root"}),
    ('level2', 'shoes', {"level1": "root", "level2": "shoes"}),
    ('level3', 'athletic shoes', {"level1": "root", "level2": "shoes", "level3": "athletic"}),
]
node_values2 = [
    ('level2', 'shirts', {"level1": "root", "level2": "shirts"}),
    ('level3', 'athletic shirts', {"level1": "root", "level2": "shirts", "level3": "athletic"}),
]

def build_branch(session, parent, node_values):
    nodes = []
    for level, name, value in node_values:
        parent = Node(level=level, name=name, parent=parent, hierarchy_hint=dumps(value))
        session.add(parent)
        nodes.append(parent)
    return nodes

with Session(engine) as session:
    nodes = []
    # Build the original branch.
    nodes.extend(build_branch(session, None, node_values))
    # Now attach another branch below root.
    nodes.extend(build_branch(session, nodes[0], node_values2))
    print(len(nodes))
    session.commit()

    # Print out to check structure
    nodes = session.query(Node).order_by(Node.id).all()
    for node in nodes:
        print(node.id, node.name, node.parent_id, node.level, node.hierarchy_hint)

    shoes = session.query(Node).filter(Node.name == 'shoes').first()
    athletic_shoes = session.query(Node).filter(Node.name == 'athletic shoes').first()
    athletic_shirts = session.query(Node).filter(Node.name == 'athletic shirts').first()
    # Check relationship...
    assert athletic_shoes in shoes.level_threes, "Athletic shoes should be below shoes"
    assert athletic_shirts not in shoes.level_threes, "Athletic shirts should not be below shoes"

输出

1 None level1 {"level1": "root"}
2 1 level2 {"level1": "root", "level2": "shoes"}
3 2 level3 {"level1": "root", "level2": "shoes", "level3": "athletic"}
4 3 level2 {"level1": "root", "level2": "shirts"}
5 4 level3 {"level1": "root", "level2": "shirts", "level3": "athletic"}