forked from cleveragents/cleveragents-core
fix(resource-registry): implement get_parents() on ResourceDagMixin
Add get_parents(name_or_id: str) -> list[Resource] method to ResourceDagMixin in _resource_registry_dag.py. The method is symmetric to the existing get_children() method but queries ResourceLinkModel by child_id instead of parent_id to return all direct parent resources. Implementation details: - Calls show_resource() first to raise NotFoundError for non-existent resources - Queries ResourceLinkModel filtered by child_id=resource.resource_id - Returns deterministically sorted list (by name, then resource_id) - Exposed via ResourceRegistryService through the mixin inheritance chain The BDD scenario 'Get parents returns all direct parents' in resource_dag.feature now passes. All 15 existing @dag_traversal and other DAG scenarios continue to pass with no regression. ISSUES CLOSED: #2844
This commit is contained in:
@@ -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