Implement path serialization

server/app/adapter/path_serialization.py

@@ -0,0 +1,63 @@
+# Copyright (c) 2026 the Eclipse BaSyx Authors
+#
+# This program and the accompanying materials are made available under the terms of the MIT License, available in
+# the LICENSE file of this project.
+#
+# SPDX-License-Identifier: MIT
+"""
+Implements the "Path" SerializationModifier
+"""
+from typing import List
+
+from basyx.aas import model
+
+
+def compute_id_short_paths(referable: model.Referable, deep: bool, include_self: bool) -> List[str]:
+    """
+    Computes the list of idShortPaths for ``referable``, as returned by the ``$path`` content modifier.
+
+    :param referable: The requested element to compute idShortPaths for.
+    :param deep: If True, all descendants on all sublevels are included (default). If False, only the requested element
+     and its direct children are included.
+    :param include_self: Whether the idShortPath of ``referable`` itself is included as the first entry.
+        This should be false for Identifiables (e.g. Submodel), since idShortPaths are not defined for them
+        as their idShort is optional.
+    :return: The list of idShortPaths
+    """
+    paths: List[str] = []
+    prefix: List[str] = []
+    if include_self:
+        if referable.id_short is None:
+            raise ValueError(f"{referable!r} has no idShort, so no idShortPath can be computed for it!")
+        prefix = [referable.id_short]
+        paths.append(model.Referable.build_id_short_path(prefix))
+    _append_child_paths(paths, referable, prefix, deep)
+    return paths
+
+
+def _append_child_paths(paths: List[str], element: object, prefix: List[str], deep: bool) -> None:
+    if isinstance(element, model.SubmodelElementList):
+        for index, child in enumerate(element.value):
+            child_prefix = prefix + [str(index)]
+            paths.append(model.Referable.build_id_short_path(child_prefix))
+            if deep:
+                _append_child_paths(paths, child, child_prefix, deep)
+        return
+
+    if isinstance(element, model.Submodel):
+        children: model.NamespaceSet = element.submodel_element
+    elif isinstance(element, model.SubmodelElementCollection):
+        children = element.value
+    elif isinstance(element, model.Entity):
+        children = element.statement
+    elif isinstance(element, model.AnnotatedRelationshipElement):
+        children = element.annotation
+    else:
+        # Leaf elements dont have children to traverse into.
+        return
+
+    for child in children:
+        child_prefix = prefix + [child.id_short]
+        paths.append(model.Referable.build_id_short_path(child_prefix))
+        if deep:
+            _append_child_paths(paths, child, child_prefix, deep)

server/app/interfaces/base.py

@@ -184,6 +184,14 @@ def __init__(self, *args, content_type="text/xml", **kwargs):
         super().__init__(*args, **kwargs, content_type=content_type)
 
 
+def assert_json_response(response_t: Type[APIResponse], context: str) -> None:
+    """
+    Raises NotAcceptable unless ``response_t`` is JsonResponse.
+    """
+    if response_t is not JsonResponse:
+        raise werkzeug.exceptions.NotAcceptable(f"{context} is only supported for the content type application/json!")
+
+
 class ResultToJsonEncoder(ServerAASToJsonEncoder):
     @classmethod
     def _result_to_json(cls, result: Result) -> Dict[str, object]:
@@ -490,3 +498,7 @@ def is_stripped_request(request: Request) -> bool:
     if extent is not None:
         raise werkzeug.exceptions.NotImplemented("The parameter extent is not yet implemented for this server!")
     return level == "core"
+
+
+def is_deep_request(request: Request) -> bool:
+    return not is_stripped_request(request)

server/app/interfaces/repository.py

@@ -23,11 +23,20 @@
 from werkzeug.routing import MapAdapter, Rule, Submount
 
 from app._config import API_BASE_PATH
+from app.adapter.path_serialization import compute_id_short_paths
 from app.interfaces.base import PagingMetadata
 from app.model import ServiceDescription, ServiceSpecificationProfileEnum
 from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter, base64url_decode
 
-from .base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, T, is_stripped_request
+from .base import (
+    APIResponse,
+    HTTPApiDecoder,
+    ObjectStoreWSGIApp,
+    T,
+    assert_json_response,
+    is_deep_request,
+    is_stripped_request,
+)
 
 SUPPORTED_PROFILES: ServiceDescription = ServiceDescription(
     [
@@ -124,7 +133,7 @@ def __init__(
                                 Rule("/$metadata", methods=["GET"], endpoint=self.get_submodel_all_metadata),
                                 Rule("/$reference", methods=["GET"], endpoint=self.get_submodel_all_reference),
                                 Rule("/$value", methods=["GET"], endpoint=self.not_implemented),
-                                Rule("/$path", methods=["GET"], endpoint=self.not_implemented),
+                                Rule("/$path", methods=["GET"], endpoint=self.get_submodel_all_path),
                                 Rule("/<base64url:submodel_id>", methods=["GET"], endpoint=self.get_submodel),
                                 Rule("/<base64url:submodel_id>", methods=["PUT"], endpoint=self.put_submodel),
                                 Rule("/<base64url:submodel_id>", methods=["DELETE"], endpoint=self.delete_submodel),
@@ -137,7 +146,7 @@ def __init__(
                                         Rule("/$value", methods=["GET"], endpoint=self.not_implemented),
                                         Rule("/$value", methods=["PATCH"], endpoint=self.not_implemented),
                                         Rule("/$reference", methods=["GET"], endpoint=self.get_submodels_reference),
-                                        Rule("/$path", methods=["GET"], endpoint=self.not_implemented),
+                                        Rule("/$path", methods=["GET"], endpoint=self.get_submodels_path),
                                         Rule(
                                             "/submodel-elements",
                                             methods=["GET"],
@@ -162,7 +171,11 @@ def __init__(
                                                     endpoint=self.get_submodel_submodel_elements_reference,
                                                 ),
                                                 Rule("/$value", methods=["GET"], endpoint=self.not_implemented),
-                                                Rule("/$path", methods=["GET"], endpoint=self.not_implemented),
+                                                Rule(
+                                                    "/$path",
+                                                    methods=["GET"],
+                                                    endpoint=self.get_submodel_submodel_elements_path,
+                                                ),
                                                 Rule(
                                                     "/<id_short_path:id_shorts>",
                                                     methods=["GET"],
@@ -210,7 +223,11 @@ def __init__(
                                                         Rule(
                                                             "/$value", methods=["PATCH"], endpoint=self.not_implemented
                                                         ),
-                                                        Rule("/$path", methods=["GET"], endpoint=self.not_implemented),
+                                                        Rule(
+                                                            "/$path",
+                                                            methods=["GET"],
+                                                            endpoint=self.get_submodel_submodel_elements_id_short_path_path,
+                                                        ),
                                                         Rule(
                                                             "/attachment",
                                                             methods=["GET"],
@@ -407,6 +424,11 @@ def _expect_namespace(cls, obj: object, needle: str) -> model.UniqueIdShortNames
             raise BadRequest(f"{obj!r} is not a namespace, can't locate {needle}!")
         return obj
 
+    @staticmethod
+    def _assert_path_supported(response_t: Type[APIResponse]) -> None:
+        # The "Path" content modifier (idShortPath serialization) is only defined for JSON in the specification.
+        assert_json_response(response_t, "The content modifier $path")
+
     @classmethod
     def _namespace_submodel_element_op(
         cls, namespace: model.UniqueIdShortNamespace, op: Callable[[str], T], arg: str
@@ -697,6 +719,16 @@ def get_submodel_all_reference(
         ]
         return response_t(references, paging_metadata=paging_metadata, stripped=is_stripped_request(request))
 
+    def get_submodel_all_path(
+        self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
+    ) -> Response:
+        self._assert_path_supported(response_t)
+        submodels, paging_metadata = self._get_submodels(request)
+        paths = [compute_id_short_paths(submodel,
+                                        is_deep_request(request),
+                                        include_self=False) for submodel in submodels]
+        return response_t(paths, paging_metadata=paging_metadata)
+
     # --------- SUBMODEL ROUTES ---------
 
     def delete_submodel(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:
@@ -722,6 +754,14 @@ def get_submodels_reference(
         reference = model.ModelReference.from_referable(submodel)
         return response_t(reference, stripped=is_stripped_request(request))
 
+    def get_submodels_path(
+        self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
+    ) -> Response:
+        self._assert_path_supported(response_t)
+        submodel = self._get_submodel(url_args)
+        paths = compute_id_short_paths(submodel, is_deep_request(request), include_self=False)
+        return response_t(paths)
+
     def put_submodel(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:
         submodel = self._get_submodel(url_args)
         submodel.update_from(HTTPApiDecoder.request_body(request, model.Submodel, is_stripped_request(request)))
@@ -753,6 +793,18 @@ def get_submodel_submodel_elements_reference(
         ]
         return response_t(references, paging_metadata=paging_metadata, stripped=is_stripped_request(request))
 
+    def get_submodel_submodel_elements_path(
+        self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
+    ) -> Response:
+        self._assert_path_supported(response_t)
+        submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args)
+        paths = [
+            compute_id_short_paths(element,
+                                   is_deep_request(request),
+                                   include_self=True) for element in submodel_elements
+        ]
+        return response_t(paths, paging_metadata=paging_metadata)
+
     def get_submodel_submodel_elements_id_short_path(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
@@ -776,6 +828,14 @@ def get_submodel_submodel_elements_id_short_path_reference(
         reference = model.ModelReference.from_referable(submodel_element)
         return response_t(reference, stripped=is_stripped_request(request))
 
+    def get_submodel_submodel_elements_id_short_path_path(
+        self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
+    ) -> Response:
+        self._assert_path_supported(response_t)
+        submodel_element = self._get_submodel_submodel_elements_id_short_path(url_args)
+        paths = compute_id_short_paths(submodel_element, is_deep_request(request), include_self=True)
+        return response_t(paths)
+
     def post_submodel_submodel_elements_id_short_path(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], map_adapter: MapAdapter
     ):