Merge pull request #29 from sachinshaji/master

Introduce session and re-try for all api responses

sw360/attachments.py

@@ -219,13 +219,14 @@ def _upload_resource_attachment(self, resource_type: str, resource_id: str, uplo
                 "application/json",
             ),
         }
-        response = requests.post(url, headers=self.api_headers, files=file_data)  # type: ignore
-        if response.status_code == HTTPStatus.ACCEPTED:
-            logger.warning(
-                f"Attachment upload was accepted by {url} but might not be visible yet: {response.text}"
-            )
-        if not response.ok:
-            raise SW360Error(response, url)
+        response = self.api_post_multipart(url, files=file_data)
+        if response is not None:
+            if response.status_code == HTTPStatus.ACCEPTED:
+                logger.warning(
+                    f"Attachment upload was accepted by {url} but might not be visible yet: {response.text}"
+                )
+            if not response.ok:
+                raise SW360Error(response, url)
 
     def upload_release_attachment(self, release_id: str, upload_file: str, upload_type: str = "SOURCE",
                                   upload_comment: str = "") -> None:

sw360/base.py

@@ -7,7 +7,7 @@
 # SPDX-License-Identifier: MIT
 # -------------------------------------------------------------------------------
 
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Dict, Optional, Tuple, Union, List
 
 import requests
 
@@ -73,6 +73,122 @@ def api_get(self, url: str = "") -> Optional[Dict[str, Any]]:
 
         raise SW360Error(response, url)
 
+    def api_post_multipart(self, url: str = "", files: Dict[str, Any] = {}) -> Optional[requests.Response]:
+        """
+        Send a multipart POST request to the specified URL with the provided file data.
+
+        :param url: The URL to send the multipart POST request to.
+        :type url: str
+        :param files: The dictionary containing file data to be sent in the request.
+        :type files: Dict[str, Any]
+        :return: The JSON response received from the server, if any.
+        :rtype: Optional[Dict[str, Any]]
+        :raises SW360Error: If the HTTP response indicates an error.
+        """
+
+        if (not self.force_no_session) and self.session is None:
+            raise SW360Error(message="login_api needs to be called first")
+
+        if self.force_no_session:
+            response = requests.post(url, headers=self.api_headers, files=files)
+        else:
+            if self.session:
+                response = self.session.post(url, files=files)
+
+        if response.ok:
+            if response.status_code == 204:  # 204 = no content
+                return None
+            return response
+
+        raise SW360Error(response, url)
+
+    def api_post(
+        self,
+        url: str = "",
+        json: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None
+    ) -> Optional[requests.Response]:
+
+        """
+        Send a POST request to the specified URL with the provided json data.
+
+        :param url: The URL to send the POST request to.
+        :type url: str
+        :param json: The dictionary containing json data to be sent in the request.
+        :type json: Dict[str, Any]
+        :return: The JSON response received from the server, if any.
+        :rtype: Optional[Dict[str, Any]]
+        :raises SW360Error: If the HTTP response indicates an error.
+        """
+
+        if (not self.force_no_session) and self.session is None:
+            raise SW360Error(message="login_api needs to be called first")
+
+        if self.force_no_session:
+            response = requests.post(url, headers=self.api_headers, json=json)
+        else:
+            if self.session:
+                response = self.session.post(url, json=json)
+
+        if response.ok:
+            if response.status_code == 204:  # 204 = no content
+                return None
+            return response
+
+        raise SW360Error(response, url)
+
+    def api_patch(self, url: str = "", json: Dict[str, Any] = {}) -> Optional[Dict[str, Any]]:
+        """
+        Send a PATCH request to the specified URL with the provided json data.
+
+        :param url: The URL to send the PATCH request to.
+        :type url: str
+        :param json: The dictionary containing json data to be sent in the request.
+        :type json: Dict[str, Any]
+        :return: The JSON response received from the server, if any.
+        :rtype: Optional[Dict[str, Any]]
+        :raises SW360Error: If the HTTP response indicates an error.
+        """
+        if (not self.force_no_session) and self.session is None:
+            raise SW360Error(message="login_api needs to be called first")
+
+        if self.force_no_session:
+            response = requests.patch(url, headers=self.api_headers, json=json)
+        else:
+            if self.session:
+                response = self.session.patch(url, json=json)
+
+        if response.ok:
+            if response.status_code == 204:  # 204 = no content
+                return None
+            return response.json()
+
+        raise SW360Error(response, url)
+
+    def api_delete(self, url: str = "") -> Optional[requests.Response]:
+        """Send a DELETE request to the specified `url` of the REST API and return JSON response.
+
+        :param url: The URL to which the DELETE request will be sent.
+        :type url: str
+        :return: JSON data returned by the API, or None if the response is empty.
+        :rtype: Optional[Dict[str, Any]]
+        :raises SW360Error: If the API responds with a non-success HTTP status code.
+        """
+        if (not self.force_no_session) and self.session is None:
+            raise SW360Error(message="login_api needs to be called first")
+
+        if self.force_no_session:
+            response = requests.delete(url, headers=self.api_headers)
+        else:
+            if self.session:
+                response = self.session.delete(url)
+
+        if response.ok:
+            if response.status_code == 204:  # 204 = no content
+                return None
+            return response
+
+        raise SW360Error(response, url)
+
     # type checking: not for Python 3.8: tuple[Optional[Any], Dict[str, Dict[str, str]], bool]
     def _update_external_ids(self, current_data: Dict[str, Any], ext_id_name: str, ext_id_value: str,
                              update_mode: str) -> Tuple[Optional[Any], Dict[str, Dict[str, str]], bool]:

sw360/components.py

@@ -11,8 +11,6 @@
 
 from typing import Any, Dict, List, Optional
 
-import requests
-
 from .base import BaseMixin
 from .sw360error import SW360Error
 
@@ -197,13 +195,12 @@ def create_new_component(self, name: str, description: str, component_type: str,
             component_details[param] = locals()[param]
         component_details["componentType"] = component_type
 
-        response = requests.post(
-            url, json=component_details, headers=self.api_headers
-        )
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        response = self.api_post(
+            url, json=component_details)
+        if response is not None:
+            if response.ok:
+                return response.json()
+        return None
 
     def update_component(self, component: Dict[str, Any], component_id: str) -> Optional[Dict[str, Any]]:
         """Update an existing component
@@ -223,14 +220,7 @@ def update_component(self, component: Dict[str, Any], component_id: str) -> Opti
             raise SW360Error(message="No component id provided!")
 
         url = self.url + "resource/api/components/" + component_id
-        response = requests.patch(
-            url, json=component, headers=self.api_headers,
-        )
-
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        return self.api_patch(url, json=component)
 
     def update_component_external_id(self, ext_id_name: str, ext_id_value: str,
                                      component_id: str, update_mode: str = "none") -> Optional[Dict[str, Any]]:
@@ -282,13 +272,11 @@ def delete_component(self, component_id: str) -> Optional[Dict[str, Any]]:
             raise SW360Error(message="No component id provided!")
 
         url = self.url + "resource/api/components/" + component_id
-        response = requests.delete(
-            url, headers=self.api_headers,
-        )
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        response = self.api_delete(url)
+        if response is not None:
+            if response.ok:
+                return response.json()
+        return None
 
     def get_users_of_component(self, component_id: str) -> Optional[Dict[str, Any]]:
         """Get information of about the users of a component

sw360/license.py

@@ -50,13 +50,13 @@ def create_new_license(
         license_details["text"] = text
         license_details["checked"] = checked
 
-        response = requests.post(url, json=license_details, headers=self.api_headers)
-        if response.ok:
-            return response.json()
-
+        response = self.api_post(url, json=license_details)
+        if response is not None:
+            if response.ok:
+                return response.json()
         raise SW360Error(response, url)
 
-    def delete_license(self, license_shortname: str) -> bool:
+    def delete_license(self, license_shortname: str) -> Optional[bool]:
         """Delete an existing license
 
         API endpoint: PATCH /licenses
@@ -73,14 +73,11 @@ def delete_license(self, license_shortname: str) -> bool:
 
         url = self.url + "resource/api/licenses/" + license_shortname
         print(url)
-        response = requests.delete(
-            url, headers=self.api_headers,
-        )
-
-        if response.ok:
-            return True
-
-        raise SW360Error(response, url)
+        response = self.api_delete(url)
+        if response is not None:
+            if response.ok:
+                return True
+        return None
 
     def download_license_info(
         self, project_id: str, filename: str, generator: str = "XhtmlGenerator", variant: str = "DISCLOSURE"

sw360/project.py

@@ -11,8 +11,6 @@
 
 from typing import Any, Dict, List, Optional
 
-import requests
-
 from .base import BaseMixin
 from .sw360error import SW360Error
 
@@ -305,13 +303,11 @@ def create_new_project(self, name: str, project_type: str, visibility: Any,
         project_details["projectType"] = project_type
 
         url = self.url + "resource/api/projects"
-        response = requests.post(
-            url, json=project_details, headers=self.api_headers
-        )
-
-        if response.ok:
-            return response.json()
-
+        response = self.api_post(
+            url, json=project_details)
+        if response is not None:
+            if response.ok:
+                return response.json()
         raise SW360Error(response, url)
 
     def update_project(self, project: Dict[str, Any], project_id: str,
@@ -345,14 +341,12 @@ def update_project(self, project: Dict[str, Any], project_id: str,
                         nsp["projectRelationship"] = sp.get("relation", "CONTAINED")
                         project["linkedProjects"][pid] = nsp
 
-        response = requests.patch(url, json=project, headers=self.api_headers)
+        return self.api_patch(url, json=project)
 
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
-
-    def update_project_releases(self, releases: List[Dict[str, Any]], project_id: str, add: bool = False) -> bool:
+    def update_project_releases(
+        self,
+        releases: List[Dict[str, Any]], project_id: str, add: bool = False
+    ) -> Optional[bool]:
         """Update the releases of an existing project. If `add` is True,
         given `releases` are added to the project, otherwise, the existing
         releases will be replaced.
@@ -383,12 +377,11 @@ def update_project_releases(self, releases: List[Dict[str, Any]], project_id: st
                 releases = old_releases + list(releases)
 
         url = self.url + "resource/api/projects/" + project_id + "/releases"
-        response = requests.post(url, json=releases, headers=self.api_headers)
-
-        if response.ok:
-            return True
-
-        raise SW360Error(response, url)
+        response = self.api_post(url, json=releases)
+        if response is not None:
+            if response.ok:
+                return True
+        return None
 
     def update_project_external_id(self, ext_id_name: str, ext_id_value: str,
                                    project_id: str, update_mode: str = "none") -> Any:
@@ -441,13 +434,11 @@ def delete_project(self, project_id: str) -> Optional[Dict[str, Any]]:
             raise SW360Error(message="No project id provided!")
 
         url = self.url + "resource/api/projects/" + project_id
-        response = requests.delete(
-            url, headers=self.api_headers
-        )
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        response = self.api_delete(url)
+        if response is not None:
+            if response.ok:
+                return response.json()
+        return None
 
     def get_users_of_project(self, project_id: str) -> Optional[Dict[str, Any]]:
         """Get information of about users of a project
@@ -486,14 +477,12 @@ def duplicate_project(self, project_id: str, new_version: str) -> Optional[Dict[
         project_details["clearingState"] = "OPEN"
 
         url = self.url + "resource/api/projects/duplicate/" + project_id
-        response = requests.post(
-            url, json=project_details, headers=self.api_headers
-        )
-
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        response = self.api_post(
+            url, json=project_details)
+        if response is not None:
+            if response.ok:
+                return response.json()
+        return None
 
     def update_project_release_relationship(
         self, project_id: str, release_id: str, new_state: str,
@@ -528,9 +517,4 @@ def update_project_release_relationship(
         relation["comment"] = comment
 
         url = self.url + "resource/api/projects/" + project_id + "/release/" + release_id
-        response = requests.patch(url, json=relation, headers=self.api_headers)
-
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        return self.api_patch(url, json=relation)