sqlalchemy / ltree 更新请求结果与预期不同
Posted
技术标签:
【中文标题】sqlalchemy / ltree 更新请求结果与预期不同【英文标题】:sqlalchemy / ltree update request result different than the expected one 【发布时间】:2022-01-03 19:22:55 【问题描述】:我在 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 的新用户,也许我忘记了什么?
提前谢谢你:-)
【问题讨论】:
【参考方案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)
【讨论】:
以上是关于sqlalchemy / ltree 更新请求结果与预期不同的主要内容,如果未能解决你的问题,请参考以下文章