From cd162488ecf33bf6ac442f0b32f63a287e6b5d71 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Sun, 5 Apr 2026 08:27:19 +0000 Subject: [PATCH] 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 --- .../services/_resource_registry_dag.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/src/cleveragents/application/services/_resource_registry_dag.py b/src/cleveragents/application/services/_resource_registry_dag.py index a99edb26..0b5144d1 100644 --- a/src/cleveragents/application/services/_resource_registry_dag.py +++ b/src/cleveragents/application/services/_resource_registry_dag.py @@ -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,