sm_implementation.py

Go to the documentation of this file.

"""Implementation of Submodel related API calls."""
 
import json
import logging
from typing import TYPE_CHECKING
 
import requests
from pydantic import BaseModel
 
if TYPE_CHECKING:
    from aas_http_client.classes.client.aas_client import AasHttpClient
 
from aas_http_client.utilities.encoder import encode_base_64
from aas_http_client.utilities.http_helper import (
    STATUS_CODE_200,
    STATUS_CODE_201,
    STATUS_CODE_204,
    STATUS_CODE_404,
    log_response,
)
 
_logger = logging.getLogger(__name__)
 
 
class SubmodelRepoImplementation(BaseModel):
    """Implementation of Submodel related API calls."""
 
    def __init__(self, client: "AasHttpClient"):
        """Initializes the SmImplementation with the given parameters."""
        self._client = client
 
        session = client.get_session()
        if session is None:
            raise ValueError(
                "HTTP session is not initialized in the client. Call 'initialize()' method of the client before creating SubmodelRegistryImplementation instance."
            )
 
        self._session: requests.Session = session
 
    # GET /submodels/{submodelIdentifier}
    def get_submodel_by_id(self, submodel_identifier: str, level: str = "", extent: str = "") -> dict | None:
        """Returns a specific Submodel.
 
        :param submodel_identifier: The Submodels unique id
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
        :return: Submodel data or None if an error occurred
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}"
 
        params: dict[str, str] = {}
        if level:
            params["level"] = level
        if extent:
            params["extent"] = extent
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # PUT /submodels/{submodelIdentifier}
    def put_submodels_by_id(self, submodel_identifier: str, request_body: dict) -> bool:
        """Updates a existing Submodel.
 
        :param submodel_identifier: The Submodels unique id
        :param request_body: Json data of the Submodel to update
        :return: True if the update was successful, False otherwise
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.put(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # DELETE /submodels/{submodelIdentifier}
    def delete_submodel_by_id(self, submodel_identifier: str) -> bool:
        """Deletes a Submodel.
 
        :param submodel_identifier: The Submodels unique id
        :return: True if the deletion was successful, False otherwise
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.delete(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # GET /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
    def get_submodel_element_by_path_submodel_repo(
        self, submodel_identifier: str, id_short_path: str, level: str = "", extent: str = ""
    ) -> dict | None:
        """Returns a specific submodel element from the Submodel at a specified path.
 
        :param submodel_identifier: The Submodels unique id
        :param id_short_path: IdShort path to the submodel element (dot-separated)
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
        :return: Submodel element data or None if an error occurred
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
 
        params: dict[str, str] = {}
        if level:
            params["level"] = level
        if extent:
            params["extent"] = extent
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # PUT /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
    def put_submodel_element_by_path_submodel_repo(self, submodel_identifier: str, id_short_path: str, request_body: dict, level: str = "") -> bool:
        """Creates or updates an existing submodel element at a specified path within submodel elements hierarchy.
 
        :param submodel_identifier: The Submodels unique id
        :param id_short_path: IdShort path to the submodel element (dot-separated)
        :param request_body: Data for the submodel element
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :return: True if the operation was successful, False otherwise
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
 
        params: dict[str, str] = {}
        if level:
            params["level"] = level
 
        self._client.set_token()
 
        try:
            response = self._session.put(url, json=request_body, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # POST /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
    def post_submodel_element_by_path_submodel_repo(
        self, submodel_identifier: str, id_short_path: str, request_body: dict, level: str = "", extent: str = ""
    ) -> dict | None:
        """Creates a new submodel element at a specified path within submodel elements hierarchy.
 
        :param submodel_identifier: The Submodels unique id
        :param id_short_path: IdShort path to the submodel element (dot-separated)
        :param request_body: Data for the new Submodel element
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
        :return: Submodel element data or None if an error occurred
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
 
        params: dict[str, str] = {}
        if level:
            params["level"] = level
        if extent:
            params["extent"] = extent
 
        self._client.set_token()
 
        try:
            response = self._session.post(url, json=request_body, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_201:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # DELETE /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}
    def delete_submodel_element_by_path_submodel_repo(self, submodel_identifier: str, id_short_path: str) -> bool:
        """Deletes a submodel element at a specified path within the submodel elements hierarchy.
 
        :param submodel_identifier: The Submodels unique id
        :param id_short_path: IdShort path to the submodel element (dot-separated)
        :return: True if the deletion was successful, False otherwise
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}"
 
        self._client.set_token()
        try:
            response = self._session.delete(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # GET /submodels
    def get_all_submodels(
        self, semantic_id: str = "", id_short: str = "", limit: int = 0, cursor: str = "", level: str = "", extent: str = ""
    ) -> dict | None:
        """Returns all Submodels.
 
        :param semantic_id: The value of the semantic id reference (UTF8-BASE64-URL-encoded)
        :param id_short: The Submodels IdShort
        :param limit: The maximum number of elements in the response array
        :param cursor: A server-generated identifier retrieved from pagingMetadata that specifies from which position the result listing should continue
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
        :return: List of Submodel data or None if an error occurred
        """
        url = f"{self._client.base_url}/submodels"
 
        params: dict[str, str] = {}
        if semantic_id:
            params["semanticId"] = semantic_id
        if id_short:
            params["idShort"] = id_short
        if limit:
            params["limit"] = str(limit)
        if cursor:
            params["cursor"] = cursor
        if level:
            params["level"] = level
        if extent:
            params["extent"] = extent
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # POST /submodels
    def post_submodel(self, request_body: dict) -> dict | None:
        """Creates a new Submodel.
 
        :param request_body: Json data of the Submodel to post
        :return: Submodel data or None if an error occurred
        """
        url = f"{self._client.base_url}/submodels"
 
        self._client.set_token()
 
        try:
            response = self._session.post(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_201:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # GET /submodels/{submodelIdentifier}/submodel-elements
    def get_all_submodel_elements_submodel_repository(
        self, submodel_identifier: str, limit: int = 100, cursor: str = "", level: str = "", extent: str = ""
    ) -> dict | None:
        """Returns all submodel elements including their hierarchy.
 
        :param submodel_identifier: The Submodels unique id
        :param limit: The maximum number of elements in the response array
        :param cursor: A server-generated identifier retrieved from pagingMetadata that specifies from which position the result listing should continue
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
        :return: List of Submodel element data or None if an error occurred
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements"
 
        params: dict[str, str] = {}
        if limit:
            params["limit"] = str(limit)
        if cursor:
            params["cursor"] = cursor
        if level:
            params["level"] = level
        if extent:
            params["extent"] = extent
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # POST /submodels/{submodelIdentifier}/submodel-elements
    def post_submodel_element_submodel_repo(self, submodel_identifier: str, request_body: dict) -> dict | None:
        """Creates a new submodel element.
 
        :param submodel_identifier: The Submodels unique id
        :param request_body: Data for the new Submodel element
        :return: Submodel element data or None if an error occurred
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements"
 
        self._client.set_token()
 
        try:
            response = self._session.post(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_201:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # POST /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}/invoke
    def invoke_operation_submodel_repo(self, submodel_identifier: str, id_short_path: str, request_body: dict, async_: str = "async") -> dict | None:
        """Synchronously invokes an Operation at a specified path.
 
        :param submodel_identifier: The Submodels unique id
        :param id_short_path: IdShort path to the operation element (dot-separated)
        :param request_body: Input parameters for the operation
        :param async_: Determines whether an operation invocation is performed asynchronously or synchronously
        :return: Operation result or None if an error occurred
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}/invoke"
 
        self._client.set_token()
 
        try:
            response = self._session.post(url, json=request_body, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # GET /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}/$value
    def get_submodel_element_by_path_value_only_submodel_repo(self, submodel_identifier: str, id_short_path: str) -> str | None:
        """Retrieves the value of a specific SubmodelElement.
 
        :param submodel_identifier: The Submodels unique id
        :param id_short_path: IdShort path to the submodel element (dot-separated)
        :return: Submodel element value or None if an error occurred
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}/$value"
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # PATCH /submodels/{submodelIdentifier}/submodel-elements/{idShortPath}/$value
    def patch_submodel_element_by_path_value_only_submodel_repo(
        self, submodel_identifier: str, id_short_path: str, value: str, level: str = ""
    ) -> bool:
        """Updates the value of an existing SubmodelElement.
 
        :param submodel_identifier: The Submodels unique id
        :param id_short_path: IdShort path to the submodel element (dot-separated)
        :param value: Submodel element value to update as string
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :return: True if the patch was successful, False otherwise
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/submodel-elements/{id_short_path}/$value"
 
        params: dict[str, str] = {}
        if level:
            params["level"] = level
 
        self._client.set_token()
 
        try:
            response = self._session.patch(url, json=value, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' or Submodel element with IDShort path '{id_short_path}' not found.")
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # GET /submodels/{submodelIdentifier}/$value
    def get_submodel_by_id_value_only(self, submodel_identifier: str, level: str = "", extent: str = "") -> dict | None:
        """Returns a specific Submodel in the ValueOnly representation.
 
        :param submodel_identifier: The Submodels unique id
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :param extent: Determines to which extent the resource is being serialized. Available values : withBlobValue, withoutBlobValue
        :return: Submodel value as dict or None if an error occurred
        """
        params: dict[str, str] = {}
        if level:
            params["level"] = level
        if extent:
            params["extent"] = extent
 
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/$value"
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # PATCH /submodels/{submodelIdentifier}/$value
    def patch_submodel_by_id_value_only(self, submodel_identifier: str, request_body: dict, level: str = "") -> bool:
        """Updates the values of an existing Submodel.
 
        :param submodel_identifier: The Submodels unique id
        :param request_body: Submodel values to update as dict
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :return: True if the patch was successful, False otherwise
        """
        params: dict[str, str] = {}
        if level:
            params["level"] = level
 
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/$value"
 
        self._client.set_token()
 
        try:
            response = self._session.patch(url, json=request_body, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True
 
    # GET /submodels/{submodelIdentifier}/$metadata
    def get_submodel_by_id_metadata(self, submodel_identifier: str, level: str = "") -> dict | None:
        """Returns the metadata attributes of a specific Submodel.
 
        :param submodel_identifier: The Submodels unique id
        :param level: Determines the structural depth of the respective resource content. Available values : deep, core
        :return: Metadata attributes of the Submodel as dict or None if an error occurred
        """
        params: dict[str, str] = {}
        if level:
            params["level"] = level
 
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}/$metadata"
 
        self._client.set_token()
 
        try:
            response = self._session.get(url, params=params, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
                _logger.debug(response.text)
                return None
 
            if response.status_code != STATUS_CODE_200:
                log_response(response)
                return None
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return None
 
        content = response.content.decode("utf-8")
        return json.loads(content)
 
    # not supported by Java Server
 
    # PATCH /submodels/{submodelIdentifier}
    def patch_submodel_by_id(self, submodel_identifier: str, submodel_data: dict) -> bool:
        """Updates an existing Submodel.
 
        :param submodel_identifier: The Submodels unique id
        :return: True if the patch was successful, False otherwise
        """
        if not self._client.encoded_ids:
            submodel_identifier = encode_base_64(submodel_identifier)
 
        url = f"{self._client.base_url}/submodels/{submodel_identifier}"
 
        self._client.set_token()
 
        try:
            response = self._session.patch(url, json=submodel_data, timeout=self._client.time_out)
            _logger.debug(f"Call REST API url '{response.url}'")
 
            if response.status_code == STATUS_CODE_404:
                _logger.warning(f"Submodel with id '{submodel_identifier}' not found.")
                _logger.debug(response.text)
                return False
 
            if response.status_code != STATUS_CODE_204:
                log_response(response)
                return False
 
        except requests.exceptions.RequestException as e:
            _logger.error(f"Error call REST API: {e}")
            return False
 
        return True