sqlalchemy / ltree 更新请求结果与预期不同

Question

我在 session.execute 查询期间遇到了一些 sqlalchemy 问题。

sqlalchemy version       == 1.3.5
sqlalchemy utils version == 0.34.1
postgres version         == 10

我实现了一个用 sqlalchemy 更新 ltree 节点的函数，灵感来自这篇文章： https://dzone.com/articles/manipulating-trees-using-sql-and-the-postgres-ltre

我正在尝试将分支从 1 个父级移到 0 个。

           root                              parent                root
             |                                  |
        root.parent            TO          parent.child
             |
     root.parent.child

我实现了应该涵盖所有场景的函数set_ltree_path

from sqlalchemy import exc
from sqlalchemy_utils import Ltree

from api_messages import MESSAGES


def uuid_to_path(obj_uuid):
    return str(obj_uuid).replace("-", "_")


def move_to(db_object, old_path, new_path, session):
    db_object.path = new_path
    update_descendants_query = f"""
    UPDATE {db_object.__tablename__}
        SET path = :new_path || subpath(path, nlevel(:old_path) - 1)
        WHERE path <@ :old_path;
    """
    session.execute(
        update_descendants_query, {"new_path": str(new_path), "old_path": str(old_path)}
    )


def get_new_parent(db_object, parent_uuid, session):
    parent_not_found_error = MESSAGES["NOT_FOUND_IN_DATABASE"].format(
        "parent_uuid", str(parent_uuid)
    )
    try:
        new_parent = session.query(db_object.__class__).get(str(parent_uuid))
        if new_parent is None:
            raise Exception(parent_not_found_error)
        return new_parent
    except exc.SQLAlchemyError:
        raise Exception(parent_not_found_error)


def set_ltree_path(db_object, parent_uuid, session):
    old_parent_uuid = db_object.parent.uuid if db_object.parent else None

    # the element has neither old nor new parent
    if old_parent_uuid is None and parent_uuid is None:
        db_object.path = Ltree(uuid_to_path(db_object.uuid))
        return

    # the element parent hasn't change
    if str(old_parent_uuid) == str(parent_uuid):
        return

    old_path = (
        Ltree(str(db_object.path))
        if db_object.path
        else Ltree(uuid_to_path(db_object.uuid))
    )

    # the element no longer has a parent
    if parent_uuid is None:
        new_path = Ltree(uuid_to_path(db_object.uuid))
        move_to(db_object, old_path, new_path, session)
        return

    new_parent = get_new_parent(db_object, parent_uuid, session)
    new_path = Ltree(str(new_parent.path)) + uuid_to_path(db_object.uuid)

    move_to(db_object, old_path, new_path, session)

并使用db object、None 调用它，因为父节点将是根节点，而db session。 最后，父级将有正确的路径，但子级，而不是预期的parent.child 路径有一个parent.parent.child 路径。 当我尝试将更新请求发送到 postgres 时，一切正常。 我是 sql alchemy 的新用户，也许我忘记了什么？ 提前谢谢你:-)

Answer 1

我发现了问题。当我调用move_to函数时，new_path的值不正确，我只需要新分支的路径，而不是我把新分支的路径+项目id

这是该函数的新版本，它还考虑了子节点成为其父节点或祖先节点的父节点，或者节点试图成为它自己的父节点的场景

# coding=utf-8
"""
Ltree implementation inispired by this article
https://dzone.com/articles/manipulating-trees-using-sql-and-the-postgres-ltre
"""

from sqlalchemy import exc
from sqlalchemy_utils import Ltree


def uuid_to_path(obj_uuid):
    return str(obj_uuid).replace("-", "_")


def move_to(session, tablename, old_path, parent_path=None):
    update_descendants_query = f"""
    UPDATE {tablename}
        SET path = :new_path || subpath(path, nlevel(:old_path) - 1)
        WHERE path <@ :old_path;
    """
    session.execute(
        update_descendants_query,
        {"new_path": str(parent_path or ""), "old_path": str(old_path)},
    )


def get_new_parent(db_object, parent_uuid, session):
    parent_not_found_error = "parent not found"
    try:
        new_parent = session.query(db_object.__class__).get(str(parent_uuid))
        if new_parent is None:
            raise Exception(parent_not_found_error)
        return new_parent
    except exc.SQLAlchemyError:
        raise Exception(parent_not_found_error)


def set_ltree_path(db_object, parent_uuid, session):
    old_parent_uuid = str(db_object.parent.uuid) if db_object.parent else None
    parent_uuid = str(parent_uuid) if parent_uuid else None
    child_uuid = str(db_object.uuid)
    child_path = str(db_object.path)

    # the element has neither old nor new parent
    if old_parent_uuid is None and parent_uuid is None:
        db_object.path = Ltree(uuid_to_path(child_uuid))
        return

    # the element parent hasn't change
    if old_parent_uuid == parent_uuid:
        return

    old_path = Ltree(child_path) if db_object.path else Ltree(uuid_to_path(child_uuid))

    # the element no longer has a parent
    if parent_uuid is None:
        new_path = Ltree(uuid_to_path(child_uuid))
        db_object.path = new_path
        move_to(session, db_object.__tablename__, old_path)
        return

    if parent_uuid == child_uuid:
        raise Exception("a node can't be the parent of himself")

    new_parent = get_new_parent(db_object, parent_uuid, session)

    if uuid_to_path(child_uuid) in str(new_parent.path):
        set_ltree_path(
            new_parent,
            db_object.parent.uuid if db_object.parent.uuid else None,
            session,
        )
        session.refresh(new_parent)

    new_parent_path = Ltree(str(new_parent.path))
    new_path = new_parent_path + uuid_to_path(child_uuid)
    db_object.path = new_path

    move_to(session, db_object.__tablename__, old_path, new_parent_path)