feat: add stream upload method for thumbnail

aas_http_client/classes/client/implementations/shell_implementation.py

@@ -173,7 +173,6 @@ def get_thumbnail_aas_repository(self, aas_identifier: str) -> bytes | None:
 
         return response.content
 
-    # PUT /shells/{aasIdentifier}/asset-information/thumbnail
     def put_thumbnail_aas_repository(self, aas_identifier: str, file_name: str, file: Path) -> bool:
         """Creates or updates the thumbnail of the Asset Administration Shell.
 
@@ -186,6 +185,27 @@ def put_thumbnail_aas_repository(self, aas_identifier: str, file_name: str, file
             _logger.error(f"Attachment file '{file}' does not exist.")
             return False
 
+        mime_type, _ = mimetypes.guess_type(file)
+
+        with file.open("rb") as f:
+            file_octet_stream = f.read()
+
+        return self.put_thumbnail_aas_repository_stream(aas_identifier, file_name, file_octet_stream, mime_type or "application/octet-stream")
+
+    # PUT /shells/{aasIdentifier}/asset-information/thumbnail
+    def put_thumbnail_aas_repository_stream(self, aas_identifier: str, file_name: str, file_octet_stream: Any, mime_type: str) -> bool:
+        """Creates or updates the thumbnail of the Asset Administration Shell.
+
+        :param aas_identifier: The Asset Administration Shells unique id
+        :param file_name: The name of the thumbnail file
+        :param file_octet_stream: The octet stream of the thumbnail file
+        :param mime_type: The MIME type of the thumbnail file (e.g., "image/png")
+        :return: True if the update was successful, False otherwise
+        """
+        if file_name is None or file_name == "" or file_octet_stream is None or mime_type is None or mime_type == "":
+            _logger.error(f"Attachment file '{file_name}' does not exist.")
+            return False
+
         if not self._client.encoded_ids:
             aas_identifier = encode_base_64(aas_identifier)
 
@@ -196,11 +216,10 @@ def put_thumbnail_aas_repository(self, aas_identifier: str, file_name: str, file
         self._client.set_token()
 
         try:
-            mime_type, _ = mimetypes.guess_type(file)
+            mime_type = mime_type or "application/octet-stream"
 
-            with file.open("rb") as f:
-                files = {"file": (file.name, f, mime_type or "application/octet-stream")}
-                response = self._session.put(url, files=files, params=params, timeout=self._client.time_out)
+            files: dict[str, tuple[str, Any, str]] = {"file": (file_name, file_octet_stream, mime_type)}
+            response = self._session.put(url, files=files, params=params, timeout=self._client.time_out)
 
             _logger.debug(f"Call REST API url '{response.url}'")
 

tests/test_client.py

@@ -1179,6 +1179,8 @@ def test_025_get_thumbnail_aas_repository(client: AasHttpClient):
     result = client.shells.get_thumbnail_aas_repository(shell_id)
     assert result is None
 
+
+
 def test_026_put_thumbnail_aas_repository(client: AasHttpClient):
     if client.shells is None:
         pytest.skip("Shells API is not available in this client")
@@ -1238,6 +1240,40 @@ def test_028_delete_thumbnail_aas_repository(client: AasHttpClient):
     get_result = client.shells.get_thumbnail_aas_repository(shell_id)
     assert get_result is None
 
+def test_028a_put_thumbnail_aas_repository_stream(client: AasHttpClient):
+    if client.shells is None:
+        pytest.skip("Shells API is not available in this client")
+
+    parsed = urlparse(client.base_url)
+    if parsed.port in PYTHON_SERVER_PORTS:
+        # NOTE: python server implementation differs
+        return
+
+    shell_id = SHELL_ID
+
+    if client.encoded_ids:
+        shell_id = encoder.encode_base_64(SHELL_ID)
+
+    filename = "Pen_Machine.png"
+    file = Path(f"./tests/test_data/{filename}").resolve()
+
+    with file.open("rb") as f:
+        file_octet_stream = f.read()
+
+    result = client.shells.put_thumbnail_aas_repository_stream(shell_id, file.name, file_octet_stream, "image/png")
+    assert result is True
+
+    get_result = client.shells.get_thumbnail_aas_repository(shell_id)
+    assert get_result is not None
+    assert len(get_result) > 0
+    assert get_result.startswith(b"\x89PNG\r\n\x1a\n")
+
+    delete_result = client.shells.delete_thumbnail_aas_repository(shell_id)
+    assert delete_result is True
+
+    get_result = client.shells.get_thumbnail_aas_repository(shell_id)
+    assert get_result is None
+
 def test_029_get_all_submodel_references_aas_repository(client: AasHttpClient):
     if client.shells is None:
         pytest.skip("Shells API is not available in this client")