server base: Add PagingMetadata to APIResponse

Previously the cursor value for the paging_metadata of a paginated
response was passed directly as `Optional[int]` to the constructor
of `APIResponse`. The presence of the `cursor` attribute also
controlled if the response contained a `paging_metadata`. This is
not conform to the [spec] which requires the `paging_metadata` to
be present in every paginated response. In case the response contains
all remaining results, the `cursor` value inside must be omitted.

This changes correct the behavior for `JSONResponse` to follow the
[spec]. As the `PagingMetadata` might carry additional information
in the future, a new class was used to pass all metadata from the
`_get_slice(...)` method to the resulting `APIResponse`.
For `XMLResponse` we mimic the previously implemented encoding of the
`cursor` value.

Fixes #519

[spec]: https://industrialdigitaltwin.io/aas-specifications/IDTA-01002/v3.1.2/http-rest-api/http-rest-api.html#pagination

Author: hpoeche

server/app/interfaces/base.py

@@ -146,31 +146,37 @@ def __init__(self, success: bool, messages: Optional[List[Message]] = None):
 ResponseData = Union[Result, object, List[object]]
 
 
+class PagingMetadata:
+    def __init__(self, cursor: Optional[str] = None):
+        self.cursor = cursor
+
+
 class APIResponse(abc.ABC, Response):
     @abc.abstractmethod
     def __init__(
-        self, obj: Optional[ResponseData] = None, cursor: Optional[int] = None, stripped: bool = False, *args, **kwargs
+            self, obj: Optional[ResponseData] = None, paging_metadata: Optional[PagingMetadata] = None,
+            stripped: bool = False, *args, **kwargs
     ):
         super().__init__(*args, **kwargs)
         if obj is None:
             self.status_code = 204
         else:
-            self.data = self.serialize(obj, cursor, stripped)
+            self.data = self.serialize(obj, paging_metadata, stripped)
 
     @abc.abstractmethod
-    def serialize(self, obj: ResponseData, cursor: Optional[int], stripped: bool) -> str:
+    def serialize(self, obj: ResponseData, paging_metadata: Optional[PagingMetadata], stripped: bool) -> str:
         pass
 
 
 class JsonResponse(APIResponse):
     def __init__(self, *args, content_type="application/json", **kwargs):
         super().__init__(*args, **kwargs, content_type=content_type)
 
-    def serialize(self, obj: ResponseData, cursor: Optional[int], stripped: bool) -> str:
-        if cursor is None:
+    def serialize(self, obj: ResponseData, paging_metadata: Optional[PagingMetadata], stripped: bool) -> str:
+        if paging_metadata is None:
             data = obj
         else:
-            data = {"paging_metadata": {"cursor": str(cursor)}, "result": obj}
+            data = {"paging_metadata": paging_metadata, "result": obj}
         return json.dumps(
             data, cls=StrippedResultToJsonEncoder if stripped else ResultToJsonEncoder, separators=(",", ":")
         )
@@ -180,10 +186,10 @@ class XmlResponse(APIResponse):
     def __init__(self, *args, content_type="application/xml", **kwargs):
         super().__init__(*args, **kwargs, content_type=content_type)
 
-    def serialize(self, obj: ResponseData, cursor: Optional[int], stripped: bool) -> str:
+    def serialize(self, obj: ResponseData, paging_metadata: Optional[PagingMetadata], stripped: bool) -> str:
         root_elem = etree.Element("response", nsmap=XML_NS_MAP)
-        if cursor is not None or not (isinstance(obj, list) and not obj):
-            root_elem.set("cursor", str(cursor))
+        if paging_metadata is not None:
+            root_elem.set("cursor", str(paging_metadata.cursor))
         if isinstance(obj, Result):
             result_elem = self.result_to_xml(obj, **XML_NS_MAP)
             for child in result_elem:
@@ -251,13 +257,22 @@ def _message_to_json(cls, message: Message) -> Dict[str, object]:
             "timestamp": message.timestamp.isoformat(),
         }
 
+    @classmethod
+    def _paging_metadata_to_json(cls, metadata: PagingMetadata) -> Dict[str, object]:
+        json_result: Dict[str, object] = dict()
+        if metadata.cursor is not None:
+            json_result["cursor"] = str(metadata.cursor)
+        return json_result
+
     def default(self, obj: object) -> object:
         if isinstance(obj, Result):
             return self._result_to_json(obj)
         if isinstance(obj, Message):
             return self._message_to_json(obj)
         if isinstance(obj, MessageType):
             return str(obj)
+        if isinstance(obj, PagingMetadata):
+            return self._paging_metadata_to_json(obj)
         return super().default(obj)
 
 
@@ -274,7 +289,7 @@ def __call__(self, environ, start_response) -> Iterable[bytes]:
         return response(environ, start_response)
 
     @classmethod
-    def _get_slice(cls, request: Request, iterator: Iterable[T]) -> Tuple[Iterator[T], Optional[int]]:
+    def _get_slice(cls, request: Request, iterator: Iterable[T]) -> Tuple[Iterator[T], Optional[PagingMetadata]]:
         limit_str = request.args.get("limit", default="100")
         cursor_str = request.args.get("cursor", default="1")
         try:
@@ -287,8 +302,14 @@ def _get_slice(cls, request: Request, iterator: Iterable[T]) -> Tuple[Iterator[T
         items = list(itertools.islice(iterator, start_index, end_index + 1))
         has_more = len(items) > limit
         paginated_slice = iter(items[:limit])
-        next_cursor = cursor + limit + 1 if has_more else None
-        return paginated_slice, next_cursor
+        next_cursor = str(cursor + limit + 1) if has_more else None
+
+        if next_cursor is not None or cursor > 0:
+            # add metadata if cursor was present in request
+            paging_metadata = PagingMetadata(cursor=next_cursor)
+        else:
+            paging_metadata = None
+        return paginated_slice, paging_metadata
 
     def handle_request(self, request: Request):
         map_adapter: MapAdapter = self.url_map.bind_to_environ(request.environ)

server/app/interfaces/registry.py

@@ -16,7 +16,7 @@
 from werkzeug.wrappers import Request, Response
 
 import app.model as server_model
-from app.interfaces.base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, is_stripped_request
+from app.interfaces.base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, is_stripped_request, PagingMetadata
 from app.model import DictDescriptorStore
 from app.util.converters import IdentifierToBase64URLConverter, base64url_decode
 
@@ -108,7 +108,7 @@ def __init__(self, object_store: model.AbstractObjectStore, base_path: str = "/a
 
     def _get_all_aas_descriptors(
         self, request: "Request"
-    ) -> Tuple[Iterator[server_model.AssetAdministrationShellDescriptor], Optional[int]]:
+    ) -> Tuple[Iterator[server_model.AssetAdministrationShellDescriptor], Optional[PagingMetadata]]:
 
         descriptors: Iterator[server_model.AssetAdministrationShellDescriptor] = self._get_all_obj_of_type(
             server_model.AssetAdministrationShellDescriptor
@@ -134,20 +134,20 @@ def _get_all_aas_descriptors(
                 raise BadRequest(f"Invalid assetType: '{asset_type}'")
             descriptors = filter(lambda desc: desc.asset_type == asset_type, descriptors)
 
-        paginated_descriptors, end_index = self._get_slice(request, descriptors)
-        return paginated_descriptors, end_index
+        paginated_descriptors, paging_metadata = self._get_slice(request, descriptors)
+        return paginated_descriptors, paging_metadata
 
     def _get_aas_descriptor(self, url_args: Dict) -> server_model.AssetAdministrationShellDescriptor:
         return self._get_obj_ts(url_args["aas_id"], server_model.AssetAdministrationShellDescriptor)
 
     def _get_all_submodel_descriptors(self, request: Request) -> Tuple[
-        Iterator[server_model.SubmodelDescriptor], Optional[int]
+        Iterator[server_model.SubmodelDescriptor], Optional[PagingMetadata]
     ]:
         submodel_descriptors: Iterator[server_model.SubmodelDescriptor] = self._get_all_obj_of_type(
             server_model.SubmodelDescriptor
         )
-        paginated_submodel_descriptors, end_index = self._get_slice(request, submodel_descriptors)
-        return paginated_submodel_descriptors, end_index
+        paginated_submodel_descriptors, paging_metadata = self._get_slice(request, submodel_descriptors)
+        return paginated_submodel_descriptors, paging_metadata
 
     def _get_submodel_descriptor(self, url_args: Dict) -> server_model.SubmodelDescriptor:
         return self._get_obj_ts(url_args["submodel_id"], server_model.SubmodelDescriptor)
@@ -170,8 +170,8 @@ def get_self_description(
     def get_all_aas_descriptors(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
-        aas_descriptors, cursor = self._get_all_aas_descriptors(request)
-        return response_t(list(aas_descriptors), cursor=cursor)
+        aas_descriptors, paging_metadata = self._get_all_aas_descriptors(request)
+        return response_t(list(aas_descriptors), paging_metadata=paging_metadata)
 
     def post_aas_descriptor(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], map_adapter: MapAdapter
@@ -301,8 +301,10 @@ def delete_submodel_descriptor_by_id_through_superpath(
     def get_all_submodel_descriptors(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
-        submodel_descriptors, cursor = self._get_all_submodel_descriptors(request)
-        return response_t(list(submodel_descriptors), cursor=cursor, stripped=is_stripped_request(request))
+        submodel_descriptors, paging_metadata = self._get_all_submodel_descriptors(request)
+        return response_t(
+            list(submodel_descriptors), paging_metadata=paging_metadata, stripped=is_stripped_request(request)
+        )
 
     def get_submodel_descriptor_by_id(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs

server/app/interfaces/repository.py

@@ -22,7 +22,7 @@
 from werkzeug.exceptions import BadRequest, Conflict, NotFound
 from werkzeug.routing import MapAdapter, Rule, Submount
 
-from app.interfaces.base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, T, is_stripped_request
+from app.interfaces.base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, T, is_stripped_request, PagingMetadata
 from app.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter, base64url_decode
 from .base import (ObjectStoreWSGIApp, APIResponse, is_stripped_request, HTTPApiDecoder, T,
                    ServiceSpecificationProfileEnum, ServiceDescription)
@@ -428,7 +428,9 @@ def _get_submodel_reference(
                 return ref
         raise NotFound(f"The AAS {aas!r} doesn't have a submodel reference to {submodel_id!r}!")
 
-    def _get_shells(self, request: Request) -> Tuple[Iterator[model.AssetAdministrationShell], Optional[int]]:
+    def _get_shells(
+            self, request: Request
+    ) -> Tuple[Iterator[model.AssetAdministrationShell], Optional[PagingMetadata]]:
         aas: Iterator[model.AssetAdministrationShell] = self._get_all_obj_of_type(model.AssetAdministrationShell)
 
         id_short = request.args.get("idShort")
@@ -471,13 +473,13 @@ def _get_shells(self, request: Request) -> Tuple[Iterator[model.AssetAdministrat
                 aas,
             )
 
-        paginated_aas, end_index = self._get_slice(request, aas)
-        return paginated_aas, end_index
+        paginated_aas, paging_metadata = self._get_slice(request, aas)
+        return paginated_aas, paging_metadata
 
     def _get_shell(self, url_args: Dict) -> model.AssetAdministrationShell:
         return self._get_obj_ts(url_args["aas_id"], model.AssetAdministrationShell)
 
-    def _get_submodels(self, request: Request) -> Tuple[Iterator[model.Submodel], Optional[int]]:
+    def _get_submodels(self, request: Request) -> Tuple[Iterator[model.Submodel], Optional[PagingMetadata]]:
         submodels: Iterator[model.Submodel] = self._get_all_obj_of_type(model.Submodel)
         id_short = request.args.get("idShort")
         if id_short is not None:
@@ -488,19 +490,19 @@ def _get_submodels(self, request: Request) -> Tuple[Iterator[model.Submodel], Op
                 semantic_id, model.Reference, False  # type: ignore[type-abstract]
             )
             submodels = filter(lambda sm: sm.semantic_id == spec_semantic_id, submodels)
-        paginated_submodels, end_index = self._get_slice(request, submodels)
-        return paginated_submodels, end_index
+        paginated_submodels, paging_metadata = self._get_slice(request, submodels)
+        return paginated_submodels, paging_metadata
 
     def _get_submodel(self, url_args: Dict) -> model.Submodel:
         return self._get_obj_ts(url_args["submodel_id"], model.Submodel)
 
     def _get_submodel_submodel_elements(
         self, request: Request, url_args: Dict
-    ) -> Tuple[Iterator[model.SubmodelElement], Optional[int]]:
+    ) -> Tuple[Iterator[model.SubmodelElement], Optional[PagingMetadata]]:
         submodel = self._get_submodel(url_args)
         paginated_submodel_elements: Iterator[model.SubmodelElement]
-        paginated_submodel_elements, end_index = self._get_slice(request, submodel.submodel_element)
-        return paginated_submodel_elements, end_index
+        paginated_submodel_elements, paging_metadata = self._get_slice(request, submodel.submodel_element)
+        return paginated_submodel_elements, paging_metadata
 
     def _get_submodel_submodel_elements_id_short_path(self, url_args: Dict) -> model.SubmodelElement:
         submodel = self._get_submodel(url_args)
@@ -523,8 +525,8 @@ def get_description(self, request: Request, url_args: Dict, response_t: Type[API
 
     # ------ AAS REPO ROUTES -------
     def get_aas_all(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:
-        aashells, cursor = self._get_shells(request)
-        return response_t(list(aashells), cursor=cursor)
+        aashells, paging_metadata = self._get_shells(request)
+        return response_t(list(aashells), paging_metadata=paging_metadata)
 
     def post_aas(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], map_adapter: MapAdapter
@@ -540,9 +542,9 @@ def post_aas(
     def get_aas_all_reference(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
-        aashells, cursor = self._get_shells(request)
+        aashells, paging_metadata = self._get_shells(request)
         references: list[model.ModelReference] = [model.ModelReference.from_referable(aas) for aas in aashells]
-        return response_t(references, cursor=cursor)
+        return response_t(references, paging_metadata=paging_metadata)
 
     # --------- AAS ROUTES ---------
     def get_aas(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:
@@ -651,8 +653,8 @@ def aas_submodel_refs_redirect(
 
     # ------ SUBMODEL REPO ROUTES -------
     def get_submodel_all(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:
-        submodels, cursor = self._get_submodels(request)
-        return response_t(list(submodels), cursor=cursor, stripped=is_stripped_request(request))
+        submodels, paging_metadata = self._get_submodels(request)
+        return response_t(list(submodels), paging_metadata=paging_metadata, stripped=is_stripped_request(request))
 
     def post_submodel(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], map_adapter: MapAdapter
@@ -669,17 +671,17 @@ def get_submodel_all_metadata(self, request: Request, url_args: Dict, response_t
                                   **_kwargs) -> Response:
         if "level" in request.args:
             raise BadRequest(f"level cannot be used when retrieving metadata!")
-        submodels, cursor = self._get_submodels(request)
-        return response_t(list(submodels), cursor=cursor, stripped=True)
+        submodels, paging_metadata = self._get_submodels(request)
+        return response_t(list(submodels), paging_metadata=paging_metadata, stripped=True)
 
     def get_submodel_all_reference(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
-        submodels, cursor = self._get_submodels(request)
+        submodels, paging_metadata = self._get_submodels(request)
         references: list[model.ModelReference] = [
             model.ModelReference.from_referable(submodel) for submodel in submodels
         ]
-        return response_t(references, cursor=cursor, stripped=is_stripped_request(request))
+        return response_t(references, paging_metadata=paging_metadata, stripped=is_stripped_request(request))
 
     # --------- SUBMODEL ROUTES ---------
 
@@ -713,24 +715,26 @@ def put_submodel(self, request: Request, url_args: Dict, response_t: Type[APIRes
     def get_submodel_submodel_elements(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
-        submodel_elements, cursor = self._get_submodel_submodel_elements(request, url_args)
-        return response_t(list(submodel_elements), cursor=cursor, stripped=is_stripped_request(request))
+        submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args)
+        return response_t(
+            list(submodel_elements), paging_metadata=paging_metadata, stripped=is_stripped_request(request)
+        )
 
     def get_submodel_submodel_elements_metadata(self, request: Request, url_args: Dict, response_t: Type[APIResponse],
                                                 **_kwargs) -> Response:
         if "level" in request.args:
             raise BadRequest(f"level cannot be used when retrieving metadata!")
-        submodel_elements, cursor = self._get_submodel_submodel_elements(request, url_args)
-        return response_t(list(submodel_elements), cursor=cursor, stripped=True)
+        submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args)
+        return response_t(list(submodel_elements), paging_metadata=paging_metadata, stripped=True)
 
     def get_submodel_submodel_elements_reference(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
-        submodel_elements, cursor = self._get_submodel_submodel_elements(request, url_args)
+        submodel_elements, paging_metadata = self._get_submodel_submodel_elements(request, url_args)
         references: list[model.ModelReference] = [
             model.ModelReference.from_referable(element) for element in list(submodel_elements)
         ]
-        return response_t(references, cursor=cursor, stripped=is_stripped_request(request))
+        return response_t(references, paging_metadata=paging_metadata, stripped=is_stripped_request(request))
 
     def get_submodel_submodel_elements_id_short_path(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs