fix(resource-registry): implement get_parents() on ResourceDagMixin #3252
@@ -226,6 +226,41 @@ class ResourceDagMixin:
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def get_parents(self: RegistryHost, name_or_id: str) -> list[Resource]:
|
||||
"""Get direct parents of a resource.
|
||||
|
||||
Args:
|
||||
name_or_id: Resource name or ULID.
|
||||
|
||||
Returns:
|
||||
List of parent ``Resource`` domain objects, sorted by name.
|
||||
|
||||
Raises:
|
||||
NotFoundError: If the resource does not exist.
|
||||
"""
|
||||
resource = self.show_resource(name_or_id)
|
||||
session = self._session()
|
||||
try:
|
||||
links = (
|
||||
session.query(ResourceLinkModel)
|
||||
.filter_by(child_id=resource.resource_id)
|
||||
.all()
|
||||
)
|
||||
parents: list[Resource] = []
|
||||
for link in links:
|
||||
parent_row = (
|
||||
session.query(ResourceModel)
|
||||
.filter_by(resource_id=str(link.parent_id))
|
||||
.first()
|
||||
)
|
||||
if parent_row is not None:
|
||||
parents.append(db_resource_to_domain(parent_row))
|
||||
# Deterministic ordering: sort by name, then resource_id
|
||||
parents.sort(key=lambda r: (r.name or "", r.resource_id))
|
||||
return parents
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def get_resource_tree(
|
||||
self: RegistryHost,
|
||||
name_or_id: str,
|
||||
|
||||
Reference in New Issue
Block a user