From 9a943c613d1add703c36bc5b3741af35dc569de1 Mon Sep 17 00:00:00 2001 From: Joshua Benning Date: Fri, 25 Sep 2026 11:37:12 +0200 Subject: [PATCH 1/4] serve/app/adapter: Implement path serialization helper methods --- server/app/adapter/path_serialization.py | 63 ++++++++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 server/app/adapter/path_serialization.py 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) From 183daf8e73daa3a050343eae3c3ad2b5c029d914 Mon Sep 17 00:00:00 2001 From: Joshua Benning Date: Fri, 25 Sep 2026 11:40:20 +0200 Subject: [PATCH 2/4] server: Implement related GET $path endpoints Additionally restrict to JSON only (followin spec) and add level query paramter --- server/app/interfaces/base.py | 19 +++++++++ server/app/interfaces/repository.py | 66 ++++++++++++++++++++++++++--- 2 files changed, 80 insertions(+), 5 deletions(-) diff --git a/server/app/interfaces/base.py b/server/app/interfaces/base.py index ac974cbad..f0bcb4623 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,14 @@ 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 parse_level(request: Request) -> bool: + """ + Parses the ``?level`` query parameter (SerializationModifier Level). + :return: True for ``level=deep`` (matches path_serialization.py), False for ``level=core``. + """ + level = request.args.get("level") + if level not in {"deep", "core", None}: + raise BadRequest(f"Level {level} is not a valid level!") + return level != "core" diff --git a/server/app/interfaces/repository.py b/server/app/interfaces/repository.py index fc7b4b0a0..f50b8c0e6 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_stripped_request, + parse_level, +) 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,14 @@ 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, parse_level(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 +752,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, parse_level(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 +791,16 @@ 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, parse_level(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 +824,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, parse_level(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 ): From f14020f68298ffd8635854498598a2d364914f61 Mon Sep 17 00:00:00 2001 From: Joshua Benning Date: Tue, 6 Oct 2026 13:43:21 +0200 Subject: [PATCH 3/4] server/app/interfaces/base: Deduplicate level logic --- server/app/interfaces/base.py | 11 ++--------- server/app/interfaces/repository.py | 10 +++++----- 2 files changed, 7 insertions(+), 14 deletions(-) diff --git a/server/app/interfaces/base.py b/server/app/interfaces/base.py index f0bcb4623..383b188b7 100644 --- a/server/app/interfaces/base.py +++ b/server/app/interfaces/base.py @@ -500,12 +500,5 @@ def is_stripped_request(request: Request) -> bool: return level == "core" -def parse_level(request: Request) -> bool: - """ - Parses the ``?level`` query parameter (SerializationModifier Level). - :return: True for ``level=deep`` (matches path_serialization.py), False for ``level=core``. - """ - level = request.args.get("level") - if level not in {"deep", "core", None}: - raise BadRequest(f"Level {level} is not a valid level!") - 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 f50b8c0e6..a0310bb72 100644 --- a/server/app/interfaces/repository.py +++ b/server/app/interfaces/repository.py @@ -35,7 +35,7 @@ T, assert_json_response, is_stripped_request, - parse_level, + is_deep_request, ) SUPPORTED_PROFILES: ServiceDescription = ServiceDescription( @@ -724,7 +724,7 @@ def get_submodel_all_path( ) -> Response: self._assert_path_supported(response_t) submodels, paging_metadata = self._get_submodels(request) - paths = [compute_id_short_paths(submodel, parse_level(request), include_self=False) for submodel in submodels] + 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 --------- @@ -757,7 +757,7 @@ def get_submodels_path( ) -> Response: self._assert_path_supported(response_t) submodel = self._get_submodel(url_args) - paths = compute_id_short_paths(submodel, parse_level(request), include_self=False) + 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: @@ -797,7 +797,7 @@ def get_submodel_submodel_elements_path( self._assert_path_supported(response_t) submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args) paths = [ - compute_id_short_paths(element, parse_level(request), include_self=True) for element in submodel_elements + 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) @@ -829,7 +829,7 @@ def get_submodel_submodel_elements_id_short_path_path( ) -> 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, parse_level(request), include_self=True) + 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( From e4c3a91222335be2e4e543b3432571462e1b2c49 Mon Sep 17 00:00:00 2001 From: Joshua Benning Date: Tue, 6 Oct 2026 13:45:35 +0200 Subject: [PATCH 4/4] server/app/interfaces/repository: Fix line lengths --- server/app/interfaces/repository.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/server/app/interfaces/repository.py b/server/app/interfaces/repository.py index a0310bb72..526faf5c9 100644 --- a/server/app/interfaces/repository.py +++ b/server/app/interfaces/repository.py @@ -34,8 +34,8 @@ ObjectStoreWSGIApp, T, assert_json_response, - is_stripped_request, is_deep_request, + is_stripped_request, ) SUPPORTED_PROFILES: ServiceDescription = ServiceDescription( @@ -724,7 +724,9 @@ def get_submodel_all_path( ) -> 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] + 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 --------- @@ -797,7 +799,9 @@ def get_submodel_submodel_elements_path( 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 + 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)