feat: introduce session and re-try for all api responses

sw360/attachments.py

@@ -219,7 +219,7 @@ 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
+        response = self.api_post_multipart(url, files=file_data)
         if response.status_code == HTTPStatus.ACCEPTED:
             logger.warning(
                 f"Attachment upload was accepted by {url} but might not be visible yet: {response.text}"

sw360/base.py

@@ -72,6 +72,117 @@ def api_get(self, url: str = "") -> Optional[Dict[str, Any]]:
             return response.json()
 
         raise SW360Error(response, url)
+
+    def api_post_multipart(self, url: str = "", files: dict[str, Any] = "") -> Optional[Dict[str, Any]]:
+        """
+        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: dict[str, Any] = "") -> Optional[Dict[str, Any]]:
+        """
+        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[Dict[str, Any]]:
+        """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,

sw360/components.py

@@ -197,14 +197,11 @@ 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
-        )
+        response = self.api_post(
+            url, json=component_details)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     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,14 +272,10 @@ 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,
-        )
+        response = self.api_delete(url)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     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,12 +50,10 @@ def create_new_license(
         license_details["text"] = text
         license_details["checked"] = checked
 
-        response = requests.post(url, json=license_details, headers=self.api_headers)
+        response = self.api_post(url, json=license_details)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     def delete_license(self, license_shortname: str) -> bool:
         """Delete an existing license
 
@@ -73,15 +71,10 @@ 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,
-        )
-
+        response = self.api_delete(url)
         if response.ok:
             return True
 
-        raise SW360Error(response, url)
-
     def download_license_info(
         self, project_id: str, filename: str, generator: str = "XhtmlGenerator", variant: str = "DISCLOSURE"
     ) -> None:

sw360/project.py

@@ -305,15 +305,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
-        )
-
+        response = self.api_post(
+            url, json=project_details)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     def update_project(self, project: Dict[str, Any], project_id: str,
                        add_subprojects: bool = False) -> Optional[Dict[str, Any]]:
         """Update an existing project
@@ -345,12 +341,7 @@ 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)
-
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        return self.api_patch(url, json=project)
 
     def update_project_releases(self, releases: List[Dict[str, Any]], project_id: str, add: bool = False) -> bool:
         """Update the releases of an existing project. If `add` is True,
@@ -383,12 +374,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)
+        response = self.api_post(url, json=releases)
 
         if response.ok:
             return True
 
-        raise SW360Error(response, url)
 
     def update_project_external_id(self, ext_id_name: str, ext_id_value: str,
                                    project_id: str, update_mode: str = "none") -> Any:
@@ -441,14 +431,10 @@ 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
-        )
+        response = self.api_delete(url)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     def get_users_of_project(self, project_id: str) -> Optional[Dict[str, Any]]:
         """Get information of about users of a project
 
@@ -486,15 +472,11 @@ 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
-        )
-
+        response = self.api_post(
+            url, json=project_details)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     def update_project_release_relationship(
         self, project_id: str, release_id: str, new_state: str,
             new_relation: str, comment: str) -> Optional[Dict[str, Any]]:
@@ -528,9 +510,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)

sw360/releases.py

@@ -155,14 +155,11 @@ def create_new_release(self, name: str, version: str, component_id: str,
         release_details["componentId"] = component_id
 
         url = self.url + "resource/api/releases"
-        response = requests.post(
-            url, json=release_details, headers=self.api_headers
-        )
+        response = self.api_post(
+            url, json=release_details)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     def update_release(self, release: Dict[str, Any], release_id: str) -> Optional[Dict[str, Any]]:
         """Update an existing release
 
@@ -181,11 +178,7 @@ def update_release(self, release: Dict[str, Any], release_id: str) -> Optional[D
             raise SW360Error(message="No release id provided!")
 
         url = self.url + "resource/api/releases/" + release_id
-        response = requests.patch(url, json=release, headers=self.api_headers)
-        if response.ok:
-            return response.json()
-
-        raise SW360Error(response, url)
+        return self.api_patch(url, json=release)
 
     def update_release_external_id(self, ext_id_name: str, ext_id_value: str,
                                    release_id: str, update_mode: str = "none") -> Optional[Dict[str, Any]]:
@@ -237,14 +230,10 @@ def delete_release(self, release_id: str) -> Optional[Dict[str, Any]]:
             raise SW360Error(message="No release id provided!")
 
         url = self.url + "resource/api/releases/" + release_id
-        response = requests.delete(
-            url, headers=self.api_headers
-        )
+        response = self.api_delete(url)
         if response.ok:
             return response.json()
 
-        raise SW360Error(response, url)
-
     def get_users_of_release(self, release_id: str) -> Optional[Dict[str, Any]]:
         """Get information of about the users of a release
 

sw360/sw360_api.py

@@ -33,9 +33,9 @@
     backoff_factor=30,
     allowed_methods=["HEAD", "GET", "OPTIONS", "POST", "PUT", "PATCH"]
 ))
-session = requests.Session()
-session.mount("http://", adapter)
-session.mount("https://", adapter)
+session_default = requests.Session()
+session_default.mount("http://", adapter)
+session_default.mount("https://", adapter)
 
 class SW360(
     AttachmentsMixin,
@@ -65,12 +65,12 @@ class SW360(
     :type oauth2: boolean
     """
 
-    def __init__(self, url: str, token: str, oauth2: bool = False) -> None:
+    def __init__(self, url: str, token: str, oauth2: bool = False, session: Optional[requests.Session] = session_default) -> None:
         """Constructor"""
         if url[-1] != "/":
             url += "/"
         self.url: str = url
-        self.session: Optional[requests.Session] = None
+        self.session: Optional[requests.Session] = session
 
         if oauth2:
             self.api_headers = {"Authorization": "Bearer " + token}
@@ -88,7 +88,6 @@ def login_api(self, token: str = "") -> bool:
         :raises SW360Error: if the login fails
         """
         if not self.force_no_session:
-            self.session = session
             self.session.headers = self.api_headers.copy()  # type: ignore
 
         url = self.url + "resource/api/"