sqlalchemy / ltree update request result different than the expected one

Question

I have some problem with sqlalchemy during a session.execute query.

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

I implemented a function to update ltree nodes with sqlalchemy, taking inspiration from this article: https://dzone.com/articles/manipulating-trees-using-sql-and-the-postgres-ltre

I'm trying to move the branch from 1 parent to 0.

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

I implemented the function set_ltree_path that should cover all scenarios

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)

and call it with the db object, None because the parent will be a root node, and the db session. At the end, the parent will have the right path, but the child, instead of the expected parent.child path has a parent.parent.child path. When I try the update request into postgres everything works fine. I'm pretty a new user of sql alchemy, maybe I forget something ? Thank you in advance :-)

Answer 1

I found the problem. When I called move_to function, the value of new_path was incorrect, I only needed the path of the new branch, instead I put the path of the new branch + the item id

This is the new version of the function, wich also takes into account the scenario where a child became a parent of its parent or ancestor, or a node trying to be it's own parent

# 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)