Implement node links endpoint

Author: edan-bainglass

aiida_restapi/models/node/__init__.py

@@ -1,5 +1,7 @@
+from .models import NodeLinks
 from .registry import NodeModelRegistry
 
 __all__ = [
+    'NodeLinks',
     'NodeModelRegistry',
 ]

aiida_restapi/models/node/models.py

@@ -0,0 +1,7 @@
+import pydantic as pdt
+from aiida.orm import Node
+
+
+class NodeLinks(Node.Model):
+    link_label: str = pdt.Field(description='The label of the link to the node.')
+    link_type: str = pdt.Field(description='The type of the link to the node.')

aiida_restapi/repository/node.py

@@ -17,7 +17,10 @@
 )
 from aiida.repository import File
 
+from aiida_restapi.common.pagination import PaginatedResults
+from aiida_restapi.common.query import QueryParams
 from aiida_restapi.common.types import NodeModelType, NodeType
+from aiida_restapi.models.node import NodeLinks
 
 from .entity import EntityRepository
 
@@ -142,6 +145,47 @@ def get_node_attributes(self, node_id: int) -> dict[str, t.Any]:
             ).first()[0],
         )
 
+    def get_node_links(
+        self,
+        node_id: int,
+        queries: QueryParams,
+        direction: t.Literal['incoming', 'outgoing'],
+    ) -> PaginatedResults[NodeLinks]:
+        """Get the incoming links of a node.
+
+        :param node_id: The id of the node to retrieve the incoming links for.
+        :param queries: The query parameters, including filters, order_by, page_size, and page.
+        :param direction: Specify whether to retrieve incoming or outgoing links.
+        :return: The paginated requested linked nodes.
+        """
+        node = self.entity_class.collection.get(pk=node_id)
+
+        start, end = (
+            queries.page_size * (queries.page - 1),
+            queries.page_size * queries.page,
+        )
+
+        if direction == 'incoming':
+            link_collection = node.base.links.get_incoming()
+        else:
+            link_collection = node.base.links.get_outgoing()
+
+        links: list[NodeLinks] = []
+        for link in link_collection.all()[start:end]:
+            link_params = link.node.serialize(minimal=True) | {
+                'link_label': link.link_label,
+                'link_type': link.link_type.value,
+            }
+            link_model = NodeLinks(**link_params)
+            links.append(link_model)
+
+        return PaginatedResults(
+            total=len(links),
+            page=queries.page,
+            page_size=queries.page_size,
+            results=links,
+        )
+
     def create_entity(
         self,
         model: NodeModelType,

aiida_restapi/routers/nodes.py

@@ -17,7 +17,7 @@
 from aiida_restapi.common.pagination import PaginatedResults
 from aiida_restapi.common.query import QueryParams, query_params
 from aiida_restapi.models.node import NodeModelRegistry
-from aiida_restapi.repository.node import NodeRepository
+from aiida_restapi.repository.node import NodeLinks, NodeRepository
 
 from .auth import UserInDB, get_current_active_user
 
@@ -281,6 +281,37 @@ async def get_node_extras(node_id: int) -> dict[str, t.Any]:
         raise HTTPException(status_code=500, detail=str(err)) from err
 
 
+@read_router.get(
+    '/nodes/{node_id}/links',
+    response_model=PaginatedResults[NodeLinks],
+    response_model_exclude_none=True,
+    response_model_exclude_unset=True,
+)
+@with_dbenv()
+async def get_node_links(
+    node_id: int,
+    queries: t.Annotated[QueryParams, Depends(query_params)],
+    direction: t.Literal['incoming', 'outgoing'] = Query(
+        description='Specify whether to retrieve incoming or outgoing links.',
+    ),
+) -> PaginatedResults[NodeLinks]:
+    """Get the incoming or outgoing links of a node.
+
+    :param node_id: The id of the node to retrieve the incoming links for.
+    :param queries: The query parameters, including filters, order_by, page_size, and page.
+    :param direction: Specify whether to retrieve incoming or outgoing links.
+    :return: The paginated requested linked nodes.
+    :raises HTTPException: 404 if the node with the given id does not exist,
+        500 for other failures during retrieval.
+    """
+    try:
+        return repository.get_node_links(node_id, queries, direction=direction)
+    except NotExistent:
+        raise HTTPException(status_code=404, detail=f'Could not find any node with id {node_id}')
+    except Exception as err:
+        raise HTTPException(status_code=500, detail=str(err)) from err
+
+
 @read_router.get('/nodes/{node_id}/download')
 @with_dbenv()
 async def download_node(