diff --git a/sw360/attachments.py b/sw360/attachments.py index 90b0464..11997fe 100644 --- a/sw360/attachments.py +++ b/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: diff --git a/sw360/base.py b/sw360/base.py index f077119..b024900 100644 --- a/sw360/base.py +++ b/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]: diff --git a/sw360/components.py b/sw360/components.py index db73022..bb317f7 100644 --- a/sw360/components.py +++ b/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 diff --git a/sw360/license.py b/sw360/license.py index fd24d38..6b845da 100644 --- a/sw360/license.py +++ b/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" diff --git a/sw360/project.py b/sw360/project.py index 283e757..725e1d5 100644 --- a/sw360/project.py +++ b/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) diff --git a/sw360/releases.py b/sw360/releases.py index 044d507..7e7f15e 100644 --- a/sw360/releases.py +++ b/sw360/releases.py @@ -11,8 +11,6 @@ from typing import Any, Dict, List, Optional -import requests - from .base import BaseMixin from .sw360error import SW360Error @@ -155,13 +153,12 @@ 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 - ) - if response.ok: - return response.json() - - raise SW360Error(response, url) + response = self.api_post( + url, json=release_details) + if response is not None: + if response.ok: + return response.json() + return None 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,13 +230,11 @@ 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 - ) - 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_release(self, release_id: str) -> Optional[Dict[str, Any]]: """Get information of about the users of a release diff --git a/sw360/sw360_api.py b/sw360/sw360_api.py index a5aeab5..5510a7f 100644 --- a/sw360/sw360_api.py +++ b/sw360/sw360_api.py @@ -33,9 +33,10 @@ 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 +66,18 @@ 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 +95,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/" diff --git a/sw360/vendor.py b/sw360/vendor.py index a4588c5..db24e5c 100644 --- a/sw360/vendor.py +++ b/sw360/vendor.py @@ -11,8 +11,6 @@ from typing import Any, Dict, List, Optional -import requests - from .base import BaseMixin from .sw360error import SW360Error @@ -66,15 +64,14 @@ def create_new_vendor(self, vendor: Dict[str, Any]) -> Dict[str, Any]: """ url = self.url + "resource/api/vendors" - response = requests.post( - url, json=vendor, headers=self.api_headers - ) - if response.ok: - return response.json() - + response = self.api_post( + url, json=vendor) + if response is not None: + if response.ok: + return response.json() raise SW360Error(response, url) - def update_vendor(self, vendor: Dict[str, Any], vendor_id: str) -> Dict[str, Any]: + def update_vendor(self, vendor: Dict[str, Any], vendor_id: str) -> Optional[Dict[str, Any]]: """Update an existing vendor API endpoint: PATCH /vendors @@ -90,13 +87,7 @@ def update_vendor(self, vendor: Dict[str, Any], vendor_id: str) -> Dict[str, Any raise SW360Error(message="No vendor id provided!") url = self.url + "resource/api/vendors/" + vendor_id - response = requests.patch( - url, json=vendor, headers=self.api_headers - ) - if response.ok: - return response.json() - - raise SW360Error(response, url) + return self.api_patch(url, json=vendor) def delete_vendor(self, vendor_id: str) -> Dict[str, Any]: """Delete an existing vendor @@ -114,10 +105,9 @@ def delete_vendor(self, vendor_id: str) -> Dict[str, Any]: raise SW360Error(message="No vendor id provided!") url = self.url + "resource/api/vendors/" + vendor_id - response = requests.delete( - url, headers=self.api_headers - ) - if response.ok: - return response.json() + response = self.api_delete(url) + if response is not None: + if response.ok: + return response.json() raise SW360Error(response, url) diff --git a/tests/test_sw360_base.py b/tests/test_sw360_base.py index 4b00790..83fe8e2 100644 --- a/tests/test_sw360_base.py +++ b/tests/test_sw360_base.py @@ -177,7 +177,7 @@ def test_api_get_no_content(self) -> None: @responses.activate def test_api_get_raw_not_logged_in(self) -> None: - lib = SW360(self.MYURL, self.MYTOKEN, False) + lib = SW360(self.MYURL, self.MYTOKEN, False, None) lib.force_no_session = False responses.add( @@ -298,7 +298,7 @@ def test_api_get_raw_error_string(self) -> None: else: self.assertEqual(404, context.exception.response.status_code) self.assertEqual("Error-String", context.exception.response.text) - + @responses.activate def test_login_server_not_responding(self) -> None: lib = SW360(self.MYURL, self.MYTOKEN, False) @@ -306,7 +306,7 @@ def test_login_server_not_responding(self) -> None: responses.add( responses.GET, url=self.MYURL + "resource/api/", - body = '{"error": "Internal Server Error", "message": "An unexpected error occurred on the server."}', + body='{"error": "Internal Server Error", "message": "An unexpected error occurred on the server."}', status=500, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, @@ -317,5 +317,6 @@ def test_login_server_not_responding(self) -> None: self.assertEqual(self.ERROR_MSG_NO_LOGIN, context.exception.message) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_sw360_projects.py b/tests/test_sw360_projects.py index b5c3a39..4eb5871 100644 --- a/tests/test_sw360_projects.py +++ b/tests/test_sw360_projects.py @@ -79,7 +79,7 @@ def test_get_project(self) -> None: @responses.activate def test_get_project_not_logged_in(self) -> None: - lib = SW360(self.MYURL, self.MYTOKEN, False) + lib = SW360(self.MYURL, self.MYTOKEN, False, None) responses.add( responses.GET,