diff --git a/server/app/adapter/path_serialization.py b/server/app/adapter/path_serialization.py new file mode 100644 index 000000000..da4d85b08 --- /dev/null +++ b/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) diff --git a/server/app/interfaces/base.py b/server/app/interfaces/base.py index ac974cbad..383b188b7 100644 --- a/server/app/interfaces/base.py +++ b/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) diff --git a/server/app/interfaces/repository.py b/server/app/interfaces/repository.py index fc7b4b0a0..526faf5c9 100644 --- a/server/app/interfaces/repository.py +++ b/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("/", methods=["GET"], endpoint=self.get_submodel), Rule("/", methods=["PUT"], endpoint=self.put_submodel), Rule("/", 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( "/", 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 ):