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

Question

我有一个 table 分层数据，其中每一行包含一个 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 获取给定 Node 在特定 level 的所有后代。该模型已经建立了自引用关系以获得直接后代 (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 table 但我对 SQL炼金术知道具体要做什么。

我也对如何使用递归查询或其他方式执行此操作的替代建议持开放态度，但我希望能够以与访问 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 项，也没有任何 level5 等项。 shoes.

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"}

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

how to establish a relationship between node and descendants at specific level with SQLAlchemy

python

sqlalchemy

别名关系

输出