Consolidate duplicated implementation of service specification (#527)

Previously we had two implementations of the service specification:
one in interfaces.base.py and the other one in
model.service_specification.py.

This removes the implementation in interfaces.base.py and adapt the
correponding usages to the one in model.service_specification.py.

Fixes #523

---------

Co-authored-by: s-heppner <iat@s-heppner.com>

Author: Ornella33

server/app/interfaces/base.py

@@ -42,70 +42,6 @@
 T = TypeVar("T")
 
 
-class ServiceSpecificationProfileEnum(str, enum.Enum):
-    """
-    Enumeration of all standardized Service Specification Profiles
-    from the AAS Part 2 API Specification (IDTA-01002-3-1).
-    Each profile is uniquely identified by its semantic URI.
-    """
-
-    # --- Asset Administration Shell (AAS) ---
-    AAS_FULL = "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellServiceSpecification/SSP-001"
-    AAS_READ = "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellServiceSpecification/SSP-002"
-
-    # --- Submodel ---
-    SUBMODEL_FULL = "https://admin-shell.io/aas/API/3/1/SubmodelServiceSpecification/SSP-001"
-    SUBMODEL_VALUE = "https://admin-shell.io/aas/API/3/1/SubmodelServiceSpecification/SSP-002"
-    SUBMODEL_READ = "https://admin-shell.io/aas/API/3/1/SubmodelServiceSpecification/SSP-003"
-
-    # --- AASX File Server ---
-    AASX_FILESERVER_FULL = "https://admin-shell.io/aas/API/3/1/AasxFileServerServiceSpecification/SSP-001"
-
-    # --- AAS Registry ---
-    AAS_REGISTRY_FULL = \
-        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-001"
-    AAS_REGISTRY_READ = \
-        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-002"
-    AAS_REGISTRY_BULK = \
-        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-003"
-
-    # --- Submodel Registry ---
-    SUBMODEL_REGISTRY_FULL = "https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-001"
-    SUBMODEL_REGISTRY_READ = "https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-002"
-    SUBMODEL_REGISTRY_BULK = "https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-003"
-
-    # --- AAS Repository ---
-    AAS_REPOSITORY_FULL = \
-        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRepositoryServiceSpecification/SSP-001"
-    AAS_REPOSITORY_READ = \
-        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRepositoryServiceSpecification/SSP-002"
-    AAS_REPOSITORY_BULK = \
-        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRepositoryServiceSpecification/SSP-003"
-
-    # --- Submodel Repository ---
-    SUBMODEL_REPOSITORY_FULL = "https://admin-shell.io/aas/API/3/1/SubmodelRepositoryServiceSpecification/SSP-001"
-    SUBMODEL_REPOSITORY_READ = "https://admin-shell.io/aas/API/3/1/SubmodelRepositoryServiceSpecification/SSP-002"
-    SUBMODEL_REPOSITORY_BULK = "https://admin-shell.io/aas/API/3/1/SubmodelRepositoryServiceSpecification/SSP-003"
-
-    # --- Concept Description Repository ---
-    CONCEPT_DESCRIPTION_REPOSITORY_FULL = \
-        "https://admin-shell.io/aas/API/3/1/ConceptDescriptionRepositoryServiceSpecification/SSP-001"
-    CONCEPT_DESCRIPTION_REPOSITORY_READ = \
-        "https://admin-shell.io/aas/API/3/1/ConceptDescriptionRepositoryServiceSpecification/SSP-002"
-    CONCEPT_DESCRIPTION_REPOSITORY_BULK = \
-        "https://admin-shell.io/aas/API/3/1/ConceptDescriptionRepositoryServiceSpecification/SSP-003"
-
-    # --- Discovery ---
-    DISCOVERY_FULL = "https://admin-shell.io/aas/API/3/1/DiscoveryServiceSpecification/SSP-001"
-    DISCOVERY_READ = "https://admin-shell.io/aas/API/3/1/DiscoveryServiceSpecification/SSP-002"
-
-
-# TODO: Maybe remove this in spite of spec? Too complicated structure
-class ServiceDescription:
-    def __init__(self, profiles: List[ServiceSpecificationProfileEnum]):
-        self.profiles: List[ServiceSpecificationProfileEnum] = profiles
-
-
 @enum.unique
 class MessageType(enum.Enum):
     UNDEFINED = enum.auto()

server/app/interfaces/discovery.py

@@ -5,7 +5,7 @@
 """
 
 import json
-from typing import Dict, List, Set
+from typing import Dict, List, Set, Type
 
 import werkzeug.exceptions
 from basyx.aas import model
@@ -14,8 +14,14 @@
 
 from app import model as server_model
 from app.adapter import jsonization
-from app.interfaces.base import BaseWSGIApp, HTTPApiDecoder
+from app.interfaces.base import BaseWSGIApp, HTTPApiDecoder, APIResponse
 from app.util.converters import IdentifierToBase64URLConverter, base64url_decode
+from app.model import ServiceSpecificationProfileEnum, ServiceDescription
+
+SUPPORTED_PROFILES: ServiceDescription = ServiceDescription([
+    ServiceSpecificationProfileEnum.DISCOVERY_FULL,
+    ServiceSpecificationProfileEnum.DISCOVERY_READ,
+])
 
 
 class DiscoveryStore:
@@ -90,6 +96,7 @@ def __init__(self, persistent_store: DiscoveryStore, base_path: str = "/api/v3.1
                 Submount(
                     base_path,
                     [
+                        Rule("/description", methods=["GET"], endpoint=self.get_description),
                         Rule(
                             "/lookup/shellsByAssetLink",
                             methods=["POST"],
@@ -122,8 +129,11 @@ def __init__(self, persistent_store: DiscoveryStore, base_path: str = "/api/v3.1
             strict_slashes=False,
         )
 
+    def get_description(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:
+        return response_t(SUPPORTED_PROFILES.to_dict())
+
     def get_all_aas_ids_by_asset_link(
-            self, request: Request, url_args: dict, response_t: type, **_kwargs
+            self, request: Request, url_args: dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
         asset_ids_param = request.args.get("assetIds", "")
         if not asset_ids_param:
@@ -154,7 +164,7 @@ def get_all_aas_ids_by_asset_link(
         return response_t(list(paginated_slice), cursor=cursor)
 
     def search_all_aas_ids_by_asset_link(
-        self, request: Request, url_args: dict, response_t: type, **_kwargs
+        self, request: Request, url_args: dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
         asset_links = HTTPApiDecoder.request_body_list(request, server_model.AssetLink, False)
         matching_aas_keys = set()
@@ -165,13 +175,15 @@ def search_all_aas_ids_by_asset_link(
         return response_t(list(paginated_slice), cursor=cursor)
 
     def get_all_specific_asset_ids_by_aas_id(
-        self, request: Request, url_args: dict, response_t: type, **_kwargs
+        self, request: Request, url_args: dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
         aas_identifier = str(url_args["aas_id"])
         asset_ids = self.persistent_store.get_all_specific_asset_ids_by_aas_id(aas_identifier)
         return response_t(asset_ids)
 
-    def post_all_asset_links_by_id(self, request: Request, url_args: dict, response_t: type, **_kwargs) -> Response:
+    def post_all_asset_links_by_id(
+        self, request: Request, url_args: dict, response_t: Type[APIResponse], **_kwargs
+    ) -> Response:
         aas_identifier = str(url_args["aas_id"])
         specific_asset_ids = HTTPApiDecoder.request_body_list(request, model.SpecificAssetId, False)
         self.persistent_store.add_specific_asset_ids_to_aas(aas_identifier, specific_asset_ids)
@@ -180,7 +192,9 @@ def post_all_asset_links_by_id(self, request: Request, url_args: dict, response_
         updated = {aas_identifier: self.persistent_store.get_all_specific_asset_ids_by_aas_id(aas_identifier)}
         return response_t(updated)
 
-    def delete_all_asset_links_by_id(self, request: Request, url_args: dict, response_t: type, **_kwargs) -> Response:
+    def delete_all_asset_links_by_id(
+        self, request: Request, url_args: dict, response_t: Type[APIResponse], **_kwargs
+    ) -> Response:
         aas_identifier = str(url_args["aas_id"])
         self.persistent_store.delete_specific_asset_ids_by_aas_id(aas_identifier)
         for key in list(self.persistent_store.asset_id_to_aas_ids.keys()):

server/app/interfaces/registry.py

@@ -17,9 +17,16 @@
 
 import app.model as server_model
 from app.interfaces.base import APIResponse, HTTPApiDecoder, ObjectStoreWSGIApp, is_stripped_request
-from app.model import DictDescriptorStore
+from app.model import DictDescriptorStore, ServiceSpecificationProfileEnum, ServiceDescription
 from app.util.converters import IdentifierToBase64URLConverter, base64url_decode
 
+SUPPORTED_PROFILES: ServiceDescription = ServiceDescription([
+    ServiceSpecificationProfileEnum.AAS_REGISTRY_FULL,
+    ServiceSpecificationProfileEnum.SUBMODEL_REGISTRY_FULL,
+    ServiceSpecificationProfileEnum.AAS_REGISTRY_READ,
+    ServiceSpecificationProfileEnum.SUBMODEL_REGISTRY_READ,
+])
+
 
 class RegistryAPI(ObjectStoreWSGIApp):
     def __init__(self, object_store: model.AbstractObjectStore, base_path: str = "/api/v3.1.1"):
@@ -156,15 +163,7 @@ def _get_submodel_descriptor(self, url_args: Dict) -> server_model.SubmodelDescr
     def get_self_description(
         self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs
     ) -> Response:
-        service_description = server_model.ServiceDescription(
-            profiles=[
-                server_model.ServiceSpecificationProfileEnum.AAS_REGISTRY_FULL,
-                server_model.ServiceSpecificationProfileEnum.AAS_REGISTRY_READ,
-                server_model.ServiceSpecificationProfileEnum.SUBMODEL_REGISTRY_FULL,
-                server_model.ServiceSpecificationProfileEnum.SUBMODEL_REGISTRY_READ,
-            ]
-        )
-        return response_t(service_description.to_dict())
+        return response_t(SUPPORTED_PROFILES.to_dict())
 
     # ------ AAS REGISTRY ROUTES -------
     def get_all_aas_descriptors(

server/app/interfaces/repository.py

@@ -22,14 +22,15 @@
 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.util.converters import IdentifierToBase64URLConverter, IdShortPathConverter, base64url_decode
-from .base import (ObjectStoreWSGIApp, APIResponse, is_stripped_request, HTTPApiDecoder, T,
-                   ServiceSpecificationProfileEnum, ServiceDescription)
+from .base import ObjectStoreWSGIApp, APIResponse, is_stripped_request, HTTPApiDecoder, T
+from app.model import ServiceSpecificationProfileEnum, ServiceDescription
 
 SUPPORTED_PROFILES: ServiceDescription = ServiceDescription([
     ServiceSpecificationProfileEnum.AAS_REPOSITORY_FULL,
     ServiceSpecificationProfileEnum.SUBMODEL_REPOSITORY_FULL,
+    ServiceSpecificationProfileEnum.AAS_REPOSITORY_READ,
+    ServiceSpecificationProfileEnum.SUBMODEL_REPOSITORY_READ,
 ])
 
 
@@ -515,11 +516,7 @@ def not_implemented(self, request: Request, url_args: Dict, **_kwargs) -> Respon
         raise werkzeug.exceptions.NotImplemented("This route is not implemented!")
 
     def get_description(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:
-        profiles = []
-        for profile in SUPPORTED_PROFILES.profiles:
-            profiles.append(profile.value)
-        description = {"profiles": profiles}
-        return response_t(description)
+        return response_t(SUPPORTED_PROFILES.to_dict())
 
     # ------ AAS REPO ROUTES -------
     def get_aas_all(self, request: Request, url_args: Dict, response_t: Type[APIResponse], **_kwargs) -> Response:

server/app/model/service_specification.py

@@ -1,20 +1,71 @@
-from enum import Enum
 from typing import List
+import enum
 
 
-class ServiceSpecificationProfileEnum(str, Enum):
-    AAS_REGISTRY_FULL = "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-001"  # noqa: E501
-    AAS_REGISTRY_READ = "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-002"  # noqa: E501
+class ServiceSpecificationProfileEnum(str, enum.Enum):
+    """
+    Enumeration of all standardized Service Specification Profiles
+    from the AAS Part 2 API Specification (IDTA-01002-3-1).
+    Each profile is uniquely identified by its semantic URI.
+    """
+
+    # --- Asset Administration Shell (AAS) ---
+    AAS_FULL = "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellServiceSpecification/SSP-001"
+    AAS_READ = "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellServiceSpecification/SSP-002"
+
+    # --- Submodel ---
+    SUBMODEL_FULL = "https://admin-shell.io/aas/API/3/1/SubmodelServiceSpecification/SSP-001"
+    SUBMODEL_VALUE = "https://admin-shell.io/aas/API/3/1/SubmodelServiceSpecification/SSP-002"
+    SUBMODEL_READ = "https://admin-shell.io/aas/API/3/1/SubmodelServiceSpecification/SSP-003"
+
+    # --- AASX File Server ---
+    AASX_FILESERVER_FULL = "https://admin-shell.io/aas/API/3/1/AasxFileServerServiceSpecification/SSP-001"
+
+    # --- AAS Registry ---
+    AAS_REGISTRY_FULL = \
+        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-001"
+    AAS_REGISTRY_READ = \
+        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-002"
+    AAS_REGISTRY_BULK = \
+        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRegistryServiceSpecification/SSP-003"
+
+    # --- Submodel Registry ---
     SUBMODEL_REGISTRY_FULL = "https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-001"
     SUBMODEL_REGISTRY_READ = "https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-002"
-    # TODO add other profiles
+    SUBMODEL_REGISTRY_BULK = "https://admin-shell.io/aas/API/3/1/SubmodelRegistryServiceSpecification/SSP-003"
+
+    # --- AAS Repository ---
+    AAS_REPOSITORY_FULL = \
+        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRepositoryServiceSpecification/SSP-001"
+    AAS_REPOSITORY_READ = \
+        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRepositoryServiceSpecification/SSP-002"
+    AAS_REPOSITORY_BULK = \
+        "https://admin-shell.io/aas/API/3/1/AssetAdministrationShellRepositoryServiceSpecification/SSP-003"
+
+    # --- Submodel Repository ---
+    SUBMODEL_REPOSITORY_FULL = "https://admin-shell.io/aas/API/3/1/SubmodelRepositoryServiceSpecification/SSP-001"
+    SUBMODEL_REPOSITORY_READ = "https://admin-shell.io/aas/API/3/1/SubmodelRepositoryServiceSpecification/SSP-002"
+    SUBMODEL_REPOSITORY_BULK = "https://admin-shell.io/aas/API/3/1/SubmodelRepositoryServiceSpecification/SSP-003"
+
+    # --- Concept Description Repository ---
+    CONCEPT_DESCRIPTION_REPOSITORY_FULL = \
+        "https://admin-shell.io/aas/API/3/1/ConceptDescriptionRepositoryServiceSpecification/SSP-001"
+    CONCEPT_DESCRIPTION_REPOSITORY_READ = \
+        "https://admin-shell.io/aas/API/3/1/ConceptDescriptionRepositoryServiceSpecification/SSP-002"
+    CONCEPT_DESCRIPTION_REPOSITORY_BULK = \
+        "https://admin-shell.io/aas/API/3/1/ConceptDescriptionRepositoryServiceSpecification/SSP-003"
+
+    # --- Discovery ---
+    DISCOVERY_FULL = "https://admin-shell.io/aas/API/3/1/DiscoveryServiceSpecification/SSP-001"
+    DISCOVERY_READ = "https://admin-shell.io/aas/API/3/1/DiscoveryServiceSpecification/SSP-002"
 
 
+# TODO: Maybe remove this in spite of spec? Too complicated structure
 class ServiceDescription:
     def __init__(self, profiles: List[ServiceSpecificationProfileEnum]):
         if not profiles:
             raise ValueError("At least one profile must be specified")
-        self.profiles = profiles
+        self.profiles: List[ServiceSpecificationProfileEnum] = profiles
 
     def to_dict(self):
         return {"profiles": [p.value for p in self.profiles]}