diff --git a/sw360/__init__.py b/sw360/__init__.py index ecd246e..0989cfd 100644 --- a/sw360/__init__.py +++ b/sw360/__init__.py @@ -9,6 +9,8 @@ __version__ = (1, 11, 2) +from .sorting import (BaseSortMixin, ProjectSortColumn, ReleaseSortColumn, + SortParam) from .sw360_api import SW360 from .sw360error import SW360Error from .sw360keycloak import SW360Keycloak @@ -18,5 +20,9 @@ "SW360", "SW360Error", "SW360OAuth2", - "SW360Keycloak" + "SW360Keycloak", + "ProjectSortColumn", + "ReleaseSortColumn", + "SortParam", + "BaseSortMixin", ] diff --git a/sw360/base.py b/sw360/base.py index 4856b3b..b2d965d 100644 --- a/sw360/base.py +++ b/sw360/base.py @@ -7,13 +7,21 @@ # SPDX-License-Identifier: MIT # ------------------------------------------------------------------------------- +import json +import logging +import os +from http import HTTPStatus from typing import Any, Dict, List, Optional, Tuple, Union from urllib.parse import urlencode import requests +from .sorting import SortParam from .sw360error import SW360Error +logger = logging.getLogger(__name__) +logger.setLevel(logging.WARNING) + class BaseMixin(): """Python interface to the Siemens SW360 platform @@ -25,16 +33,19 @@ class BaseMixin(): token_type is "Bearer" for an OAuth workflow and "Token" for tokens generated via the SW360 UI. - :param url: URL of the SW360 instance - :param token: The SW360 REST API token (the cryptic string without + :ivar url: URL of the SW360 instance + :ivar token: The SW360 REST API token (the cryptic string without "Authorization:" and `token_type`). - :param oauth2: flag indicating whether this is an OAuth2 token + :ivar oauth2: flag indicating whether this is an OAuth2 token + :ivar default_batch_size: Default size of batch to use while fetching all items from API :type url: string :type token: string :type oauth2: boolean + :type default_batch_size: int """ - def __init__(self, url: str, token: str, oauth2: bool = False) -> None: + def __init__(self, url: str, token: str, oauth2: bool = False, + default_batch_size: int = 50) -> None: """Constructor""" if url[-1] != "/": url += "/" @@ -47,6 +58,7 @@ def __init__(self, url: str, token: str, oauth2: bool = False) -> None: self.api_headers = {"Authorization": "Token " + token} self.force_no_session = False + self.default_batch_size = default_batch_size def api_get(self, url: str = "") -> Optional[Dict[str, Any]]: """Request `url` from REST API and return json answer. @@ -68,12 +80,76 @@ def api_get(self, url: str = "") -> Optional[Dict[str, Any]]: response = self.session.get(url) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response.json() raise SW360Error(response, url) + def api_get_all(self, url: str, sort: Optional[SortParam] = None, + batch: int = -1, page: int = 0, + _data: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]: + """ + Retrieve all pages of data from the specified URL. + + :param url: The URL to request data from. + :param sort: The sort parameter to sort data by. + :param page: Page number to fetch + :param batch: How many rows to fetch at a time, use + `default_batch_size` if -1 + :param _data: Internal param for aggregating data. + :return: The combined JSON data from all pages. + :rtype: Optional[Dict[str, Any]] + """ + if _data is None: + _data = {} + if batch == -1: + batch = self.default_batch_size + + paginated_url = self._add_pagination(url, page, batch, sort) + resp = self.api_get(paginated_url) + if resp is not None and 'page' in resp: + total_pages = resp['page']['totalPages'] + # Clean up meta info + if '_links' in resp: + del resp['_links'] + del resp['page'] + # Update data and get next page + _data = self.__merge_responses(_data, resp) + if page + 1 < total_pages: + return self.api_get_all(url, sort, batch, page + 1, _data) + else: + # Clean up meta info + if resp is not None and '_links' in resp: + del resp['_links'] + _data = self.__merge_responses(_data, resp) + return _data + + def __merge_responses(self, previous: Dict[str, Any], + next: Optional[Dict[str, Any]]) -> Dict[str, Any]: + """ + Function to merge previous and next response of the same API + recursively. + :param previous: The previous response data. + :param next: The next response data. + :return: The merged response data. + """ + if next is None: + return previous + for key, value in next.items(): + if (key in previous and isinstance(previous[key], dict) and + isinstance(value, dict)): + previous[key] = self.__merge_responses(previous[key], value) + elif (key in previous and isinstance(previous[key], list) and + isinstance(value, list)): + previous[key].extend(value) + elif (key in previous and isinstance(previous[key], tuple) and + isinstance(value, tuple)): + previous[key].extend(value) + else: + previous[key] = value + return previous + 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. @@ -97,7 +173,7 @@ def api_post_multipart(self, url: str = "", files: Dict[str, Any] = {}) -> Optio response = self.session.post(url, files=files) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response @@ -131,7 +207,7 @@ def api_post( response = self.session.post(url, json=json) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response @@ -159,7 +235,7 @@ def api_patch(self, url: str = "", json: Any = {}) -> Optional[Dict[str, Any]]: response = self.session.patch(url, json=json) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None if response.content: return response.json() @@ -187,7 +263,7 @@ def api_delete(self, url: str = "") -> Optional[requests.Response]: response = self.session.delete(url) if response.ok: - if response.status_code == 204: # 204 = no content + if response.status_code == HTTPStatus.NO_CONTENT: return None return response @@ -223,14 +299,37 @@ def _add_params(self, url: str, params: Dict[str, str]) -> str: query_string = urlencode(params) + if query_string == "": + return url + if "?" in url: return f"{url}&{query_string}" else: return f"{url}?{query_string}" + def _add_pagination(self, url: str, page: int, page_entries: int, + sort: Optional[SortParam] = None) -> str: + """ + Add pagination parameters to the GET request URL + :param url: URL to add params to + :param page: Page number to fetch + :param page_entries: Number of entries to fetch per request + :param sort: Sorting parameter (optional) + :return: URL with pagination parameters added + """ + + params = { + "page": str(page), + "page_entries": str(page_entries) + } + if sort is not None: + params["sort"] = str(sort) + + return self._add_params(url, params) + @classmethod def get_id_from_href(cls, href: str) -> str: - """"Extracts the identifier from the href and returns it + """Extracts the identifier from the href and returns it :param href: HAL href for a specific resource :type href: string (valid URL) @@ -241,3 +340,89 @@ def get_id_from_href(cls, href: str) -> str: pos = href.rfind("/") identifier = href[(pos + 1):] return identifier + + def _upload_resource_file(self, upload_file: str, + upload_type: str = "SOURCE", + upload_comment: str = "") -> Dict[str, str]: + """Upload `upload_file` as attachment to SW360 which can then be used + by various resources as attachment content. + `upload_type` can be: + "DOCUMENT" + "SOURCE" + "CLEARING_REPORT" + "COMPONENT_LICENSE_INFO_XML" + "SOURCE_SELF" + "BINARY" + "BINARY_SELF" + "LICENSE_AGREEMENT" + "README_OSS" + + API endpoint: POST /attachments + + :return: Returns the attachment content of the uploaded file which + can be used by other resources. + :raises SW360Error: if unable to extract content from the upload + """ + if not os.path.exists(upload_file): + raise SW360Error(message="ERROR: file not found: " + upload_file) + + filename = os.path.basename(upload_file) + url = self.url + "resource/api/attachments" + attachment_data = { + "filename": filename, + "createdComment": upload_comment, + "attachmentType": upload_type + } + + file_data = { + "files": (filename, open(upload_file, "rb"), "multipart/form-data"), + "attachment": ( + "", # dummy filename + json.dumps(attachment_data), + "application/json", + ), + } + response = self.api_post_multipart(url, files=file_data) + attachment_content = None + 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 response.status_code == HTTPStatus.OK: + r = response.json() + if '_embedded' in r and 'sw360:attachments' in r['_embedded'] \ + and len(r['_embedded']['sw360:attachments']) == 1: + content = r['_embedded']['sw360:attachments'][0] + attachment_content = { + 'attachmentContentId': content['attachmentContentId'], + 'filename': content['filename'], + 'sha1': content['sha1'], + 'attachmentType': content['attachmentType'], + 'createdComment': content['createdComment'], + 'checkStatus': content['checkStatus'] + } + if not response.ok: + raise SW360Error(response, url) + if attachment_content is None: + raise SW360Error(response, url, + "Unable to fetch attachment content from the response.") + return attachment_content + + @staticmethod + def _get_attachments( + resource: Optional[Dict[str, Any]] + ) -> List[Dict[str, Any]]: + attachments: List[Dict[str, Any]] = [] + if resource is None: + return attachments + if 'attachments' in resource: + attachments.extend(resource['attachments']) + if '_embedded' in resource and \ + 'sw360:attachments' in resource['_embedded']: + attachments.extend(resource['_embedded']['sw360:attachments']) + # Remove meta properties like "_links" from attachments + for attachment in attachments: + if '_links' in attachment: + del attachment['_links'] + return attachments diff --git a/sw360/project.py b/sw360/project.py index 6b5d44a..2d5e40e 100644 --- a/sw360/project.py +++ b/sw360/project.py @@ -12,6 +12,7 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import ProjectSortColumn, ReleaseSortColumn, SortParam from .sw360error import SW360Error @@ -19,7 +20,7 @@ class ProjectMixin(BaseMixin): def get_project(self, project_id: str) -> Optional[Dict[str, Any]]: """Get information of about a project - API endpoint: GET /projects + API endpoint: GET /projects/{id} :param project_id: the id of the project to be requested :type project_id: string @@ -52,8 +53,9 @@ def get_project_releases(self, project_id: str, transitive: bool = False) -> Any trans = "false" if transitive: trans = "true" - resp = self.api_get(self.url + "resource/api/projects/" - + project_id + "/releases?transitive=" + trans) + url = self.url + "resource/api/projects/" + project_id + "/releases" + url = self._add_params(url, {"transitive": trans}) + resp = self.api_get_all(url, ReleaseSortColumn.CREATED_ON.desc()) return resp def get_project_by_url(self, url: str) -> Optional[Dict[str, Any]]: @@ -73,20 +75,64 @@ def get_project_by_url(self, url: str) -> Optional[Dict[str, Any]]: resp = self.api_get(url) return resp - def get_projects(self, all_details: bool = False, page: int = -1, - page_size: int = -1, sort: str = "") -> Optional[Dict[str, Any]]: + def __get_projects_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: + """ + Take a pre-generated URL of projects endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + projects. + + :param url: projects API URL with filters in the query + :type url: str + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by name if `None`) + :type sort: SortParam + :return: list of projects + :rtype: list of JSON project objects + :raises SW360Error: if there is a negative HTTP response + """ + + full_url = self._add_params(url, {"luceneSearch": "true"}) + if page > -1 and page_size > -1: + full_url = self._add_pagination(url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + + if not resp: + return [] + + if "_embedded" not in resp: + return [] + + if "sw360:projects" not in resp["_embedded"]: + return [] + + return resp["_embedded"]["sw360:projects"] + + def get_projects( + self, all_details: bool = False, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get all projects API endpoint: GET /projects - :param all_details: retrieve all project details (optional)) + :param all_details: retrieve all project details (optional) :type all_details: bool :param page: page to retrieve :type page: int - :param page_size: page size to use + :param page_size: page size to use, `-1` to get all :type page_size: int - :param sort: sort order for the projects ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the projects (Sort by name if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -97,25 +143,30 @@ def get_projects(self, all_details: bool = False, page: int = -1, if all_details: params["allDetails"] = "true" + url_with_param = self._add_params(fullbase_url, params) - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + if sort is None: + sort = ProjectSortColumn.NAME.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - return resp + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) - def get_projects_by_type(self, project_type: str) -> List[Dict[str, Any]]: + def get_projects_by_type( + self, project_type: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get information of about all projects of a certain type API endpoint: GET /projects :param project_type: the full url of the project to be requested :type project_type: string, one of CUSTOMER, INTERNAL, PRODUCT, SERVICE, INNER_SOURCE + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by name if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -123,17 +174,16 @@ def get_projects_by_type(self, project_type: str) -> List[Dict[str, Any]]: if not project_type: raise SW360Error(message="No project type provided!") - resp = self.api_get(self.url + "resource/api/projects?type=" + project_type) - if not resp: - return [] + fullbase_url = self.url + "resource/api/projects" + params = {"type": project_type} - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.NAME.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) def get_project_names(self) -> List[str]: """Get all project names @@ -149,13 +199,7 @@ def get_project_names(self) -> List[str]: if not all_projects: return resp - if "_embedded" not in all_projects: - return resp - - if "sw360:projects" not in all_projects["_embedded"]: - return resp - - projects: List[Dict[str, Any]] = all_projects["_embedded"]["sw360:projects"] + projects: List[Dict[str, Any]] = all_projects if not projects: return resp @@ -164,13 +208,22 @@ def get_project_names(self) -> List[str]: return resp - def get_projects_by_name(self, name: str) -> List[Dict[str, Any]]: + def get_projects_by_name( + self, name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get a project by its name API endpoint: GET /projects :param name: the project name or a prefix of it :type name: string + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by score if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -178,19 +231,20 @@ def get_projects_by_name(self, name: str) -> List[Dict[str, Any]]: if not name: raise SW360Error(message="No project name provided!") - resp = self.api_get(self.url + "resource/api/projects?name=" + name) - if not resp: - return [] + fullbase_url = self.url + "resource/api/projects" + params = {"name": name} - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.SCORE.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) - def get_projects_by_external_id(self, ext_id_name: str, ext_id_value: str = "") -> List[Dict[str, Any]]: + def get_projects_by_external_id( + self, ext_id_name: str, ext_id_value: str = "" + ) -> List[Dict[str, Any]]: """Get projects by external id. `ext_id_value` can be left blank to search for all projects with `ext_id_name`. @@ -223,13 +277,24 @@ def get_projects_by_external_id(self, ext_id_name: str, ext_id_value: str = "") return resp["_embedded"]["sw360:projects"] - def get_projects_by_group(self, group: str, all_details: bool = False) -> List[Dict[str, Any]]: + def get_projects_by_group( + self, group: str, all_details: bool = False, page: int = -1, + page_size: int = -1, sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get projects by group. API endpoint: GET /projects?group= :param group: the group the projects shall belong to :type group: string + :param all_details: retrieve all project details (optional) + :type all_details: bool + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by score if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -237,29 +302,36 @@ def get_projects_by_group(self, group: str, all_details: bool = False) -> List[D if not group: raise SW360Error(message="No group provided!") - full_url = self.url + "resource/api/projects?group=" + group - if all_details: - full_url = self.url + "resource/api/projects?allDetails?group=" + group + fullbase_url = self.url + "resource/api/projects" + params = {"group": group} - resp = self.api_get(full_url) - if not resp: - return [] + if all_details: + params["allDetails"] = "true" - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.SCORE.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) - def get_projects_by_tag(self, tag: str) -> List[Dict[str, Any]]: + def get_projects_by_tag( + self, tag: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get projects by tag. API endpoint: GET /projects?tag= - :param group: the group the projects shall belong to - :type group: string + :param tag: the group the projects shall belong to + :type tag: string + :param page: page to retrieve + :type page: int + :param page_size: page size to use, `-1` to get all + :type page_size: int + :param sort: sort order for the projects (Sort by score if `None`) + :type sort: SortParam :return: list of projects :rtype: list of JSON project objects :raises SW360Error: if there is a negative HTTP response @@ -267,18 +339,16 @@ def get_projects_by_tag(self, tag: str) -> List[Dict[str, Any]]: if not tag: raise SW360Error(message="No tag provided!") - full_url = self.url + "resource/api/projects?tag=" + tag + "&luceneSearch=true" - resp = self.api_get(full_url) - if not resp: - return [] + fullbase_url = self.url + "resource/api/projects" + params = {"tag": tag} - if "_embedded" not in resp: - return [] + url_with_param = self._add_params(fullbase_url, params) - if "sw360:projects" not in resp["_embedded"]: - return [] + if sort is None: + sort = ProjectSortColumn.SCORE.asc() - return resp["_embedded"]["sw360:projects"] + return self.__get_projects_filtered(url_with_param, page, page_size, + sort) def get_project_vulnerabilities(self, project_id: str) -> Optional[Dict[str, Any]]: """Get the security vulnerabilities for the specified project. @@ -295,15 +365,17 @@ def get_project_vulnerabilities(self, project_id: str) -> Optional[Dict[str, Any raise SW360Error(message="No project id provided!") full_url = self.url + "resource/api/projects/" + project_id + "/vulnerabilities" - resp = self.api_get(full_url) + resp = self.api_get_all(full_url) if not resp: return None return resp - def create_new_project(self, name: str, project_type: str, visibility: Any, - description: str = "", version: str = "", - project_details: Dict[str, Any] = {}) -> Optional[Dict[str, Any]]: + def create_new_project( + self, name: str, project_type: str, visibility: Any, + description: str = "", version: str = "", + project_details: Dict[str, Any] = {} + ) -> Optional[Dict[str, Any]]: """Create a new project. The parameters list only the most common project attributes, check the @@ -341,8 +413,10 @@ def create_new_project(self, name: str, project_type: str, visibility: Any, 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]]: + def update_project( + self, project: Dict[str, Any], project_id: str, + add_subprojects: bool = False + ) -> Optional[Dict[str, Any]]: """Update an existing project API endpoint: PATCH /projects @@ -367,12 +441,15 @@ def update_project(self, project: Dict[str, Any], project_id: str, if add_subprojects: current = self.get_project(project_id) - if (current is not None and "linkedProjects" in current and "linkedProjects" in project): + if current is not None and "linkedProjects" in current and "linkedProjects" in project: for sp in current["linkedProjects"]: pid = self.get_id_from_href(sp["project"]) if pid not in project["linkedProjects"]: - nsp = {} - nsp["projectRelationship"] = sp.get("relation", "CONTAINED") + nsp = { + "projectRelationship": sp.get( + "relation", "CONTAINED" + ) + } project["linkedProjects"][pid] = nsp return self.api_patch(url, json=project) @@ -385,13 +462,13 @@ def update_project_releases( given `releases` are added to the project, otherwise, the existing releases will be replaced. - API endpoint: POST /projects//releases + API endpoint: POST /projects/{id}/releases - :param releases: list of relase_ids to be linked in the project - :param project_id: the id of the project to modify - :param add: add given releases if set to True, replace otherwise + :param releases: list of release_ids to be linked in the project :type releases: list of release_id strings + :param project_id: the id of the project to modify :type project_id: string + :param add: add given releases if set to True, replace otherwise :type add: boolean :return: SW360 result :rtype: JSON SW360 result object @@ -421,7 +498,7 @@ def update_project_external_id(self, ext_id_name: str, ext_id_value: str, The method will return the old value of the external id or None if it was not set. - API endpoint: PATCH /projects + API endpoint: PATCH /projects/{id} :param ext_id_name: name of the external id :param ext_id_value: value of the external id @@ -458,7 +535,7 @@ def update_project_external_id(self, ext_id_name: str, ext_id_value: str, def delete_project(self, project_id: str) -> Optional[Dict[str, Any]]: """Delete an existing project - API endpoint: DELETE /projects + API endpoint: DELETE /projects/{id} :param project_id: the id of the project to be requested :type project_id: string @@ -499,7 +576,7 @@ def get_users_of_project(self, project_id: str) -> Optional[Dict[str, Any]]: def duplicate_project(self, project_id: str, new_version: str) -> Optional[Dict[str, Any]]: """Create a copy of an existing project. - API endpoint: GET /projects/duplicate/{id} + API endpoint: POST /projects/duplicate/{id} :param project_id: the id of the existing project :type project_id: string @@ -534,7 +611,7 @@ def update_project_release_relationship( new_relation: str, comment: str) -> Optional[Dict[str, Any]]: """Update the relationship for a specific release of a project - API endpoint PATCH /projects/{pid}/release{rid} + API endpoint PATCH /projects/{pid}/release/{rid} :param project_id: the id of the existing project :type project_id: string @@ -573,7 +650,7 @@ def update_project_release_relationship( def link_packages_to_project(self, project_id: str, packages: List[str]) -> Optional[Dict[str, Any]]: """Link (new) packages to a given project. - API endpoint PATCH /projects/{pid}/packages{rid} + API endpoint PATCH /projects/{pid}/packages/{rid} :param project_id: the id of the existing project :type project_id: string @@ -594,7 +671,7 @@ def link_packages_to_project(self, project_id: str, packages: List[str]) -> Opti def unlink_packages_from_project(self, project_id: str, packages: List[str]) -> Optional[Dict[str, Any]]: """Unlink packages from a given project. - API endpoint PATCH /projects/{pid}/packages{rid} + API endpoint PATCH /projects/{pid}/packages/{rid} :param project_id: the id of the existing project :type project_id: string @@ -611,3 +688,35 @@ def unlink_packages_from_project(self, project_id: str, packages: List[str]) -> url = self.url + "resource/api/projects/" + project_id + "/unlink/packages/" return self.api_patch(url, json=packages) + + def upload_attachment_to_project( + self, project_id: str, upload_file: str, upload_type: str = "SOURCE", + upload_comment: str = "" + ) -> Optional[Dict[str, Any]]: + """Upload an attachment to a given project. + + API endpoint: POST /attachments & PATCH /projects/{id} + + :param project_id: the id of the Project + :type project_id: string + :param upload_file: path of the file to be uploaded + :type upload_file: string + :param upload_type: the type of the attachment + :type upload_type: string + :param upload_comment: a comment for the attachment + :type upload_comment: string + :raises SW360Error: if the project id is missing or there is a negative HTTP response + """ + if not project_id: + raise SW360Error(message="No project id provided!") + + attachment_content = self._upload_resource_file(upload_file, upload_type, upload_comment) + # Make sure the type is correct, override + attachment_content['attachmentType'] = upload_type + attachment_content['createdComment'] = upload_comment + + current_project = self.get_project(project_id) + attachments = self._get_attachments(current_project) + + attachments.append(attachment_content) + return self.update_project({'attachments': attachments}, project_id) diff --git a/sw360/sorting.py b/sw360/sorting.py new file mode 100644 index 0000000..0494661 --- /dev/null +++ b/sw360/sorting.py @@ -0,0 +1,46 @@ +# ------------------------------------------------------------------------------ +# Copyright (c) 2026 Siemens +# All Rights Reserved. +# Authors: mishra.gaurav@siemens.com +# +# Licensed under the MIT license. +# SPDX-License-Identifier: MIT +# ------------------------------------------------------------------------------ +from enum import Enum + + +class SortParam: + def __init__(self, value: str): + self.value = value + + def __str__(self) -> str: + return self.value + + +class BaseSortMixin(Enum): + def asc(self) -> SortParam: + return SortParam(f"{self.value},asc") + + def desc(self) -> SortParam: + return SortParam(f"{self.value},desc") + + +class ProjectSortColumn(BaseSortMixin, Enum): + SCORE = "score" + CREATED_ON = "createdOn" + NAME = "name" + VENDOR = "vendor" + LICENSE = "license" + TYPE = "type" + DESCRIPTION = "description" + PROJECT_RESPONSIBLE = "projectResponsible" + STATE = "state" + + +class ReleaseSortColumn(BaseSortMixin, Enum): + CREATED_ON = "createdOn" + NAME = "name" + VERSION = "version" + CLEARING_STATE = "clearingState" + MAINLINE_STATE = "mainlineState" + SCORE = "score" diff --git a/sw360/sw360_api.py b/sw360/sw360_api.py index 795f4cf..d6450ce 100644 --- a/sw360/sw360_api.py +++ b/sw360/sw360_api.py @@ -68,6 +68,8 @@ class SW360( :type url: string :type token: string :type oauth2: boolean + :ivar default_batch_size: Default size of batch to use while fetching all items from API + :type default_batch_size: int """ def __init__( @@ -75,9 +77,11 @@ def __init__( url: str, token: str, oauth2: bool = False, - session: Optional[requests.Session] = session_default + session: Optional[requests.Session] = session_default, + default_batch_size: int = 50 ) -> None: """Constructor""" + super().__init__(url, token, oauth2, default_batch_size) if url[-1] != "/": url += "/" self.url: str = url @@ -89,6 +93,7 @@ def __init__( self.api_headers = {"Authorization": "Token " + token} self.force_no_session = False + self.default_batch_size = default_batch_size def login_api(self, token: str = "") -> bool: """Login to SW360 REST API. This used to have a `token` parameter diff --git a/tests/test_sw360_projects.py b/tests/test_sw360_projects.py index c1a59f8..38b4d53 100644 --- a/tests/test_sw360_projects.py +++ b/tests/test_sw360_projects.py @@ -20,6 +20,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import ProjectSortColumn # noqa: E402 class Sw360TestProjects(unittest.TestCase): @@ -140,7 +141,7 @@ def test_get_project_releases(self) -> None: lib = self.get_logged_in_lib() responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", body='{"name": "My Testproject"}', status=200, content_type="application/json", @@ -152,7 +153,7 @@ def test_get_project_releases(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=true", + url=self.MYURL + "resource/api/projects/123/releases?transitive=true&page=0&page_entries=50&sort=createdOn,desc", body='{"name": "My Testproject"}', status=200, content_type="application/json", @@ -185,7 +186,7 @@ def test_get_projects(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects", + url=self.MYURL + "resource/api/projects?luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -195,9 +196,7 @@ def test_get_projects(self) -> None: projects = lib.get_projects() self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_with_details(self) -> None: @@ -205,7 +204,7 @@ def test_get_projects_with_details(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?allDetails=true", + url=self.MYURL + "resource/api/projects?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -215,9 +214,7 @@ def test_get_projects_with_details(self) -> None: projects = lib.get_projects(all_details=True) self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_with_paging(self) -> None: @@ -225,7 +222,7 @@ def test_get_projects_with_paging(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?page=1&page_entries=2", + url=self.MYURL + "resource/api/projects?page=1&page_entries=2&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -235,9 +232,7 @@ def test_get_projects_with_paging(self) -> None: projects = lib.get_projects(page=1, page_size=2) self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_with_paging_and_details(self) -> None: @@ -252,12 +247,10 @@ def test_get_projects_with_paging_and_details(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - projects = lib.get_projects(all_details=True, page=3, page_size=4, sort="name,desc") + projects = lib.get_projects(all_details=True, page=3, page_size=4, sort=ProjectSortColumn.NAME.desc()) self.assertIsNotNone(projects) if projects: # only for mypy - self.assertTrue("_embedded" in projects) - self.assertTrue("sw360:projects" in projects["_embedded"]) - self.assertEqual("My Testproject", projects["_embedded"]["sw360:projects"][0]["name"]) + self.assertEqual("My Testproject", projects[0]["name"]) @responses.activate def test_get_projects_by_type(self) -> None: @@ -265,7 +258,7 @@ def test_get_projects_by_type(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "projectType": "SERVICE"}]}}', # noqa status=200, content_type="application/json", @@ -284,7 +277,7 @@ def test_get_projects_by_type_no_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{}', status=200, content_type="application/json", @@ -301,7 +294,7 @@ def test_get_projects_by_type_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject", "projectType": "SERVICE"}]}}', # noqa status=200, content_type="application/json", @@ -318,7 +311,7 @@ def test_get_projects_by_type_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?type=SERVICE", + url=self.MYURL + "resource/api/projects?type=SERVICE&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject", "projectType": "SERVICE"}]}}', # noqa status=200, content_type="application/json", @@ -335,7 +328,7 @@ def test_get_project_names(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects", + url=self.MYURL + "resource/api/projects?luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "version" : "1.0.0"}]}}', # noqa status=200, content_type="application/json", @@ -404,7 +397,7 @@ def test_get_projects_by_name(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{}', status=200, content_type="application/json", @@ -420,7 +413,7 @@ def test_get_projects_by_name_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -438,7 +431,7 @@ def test_get_projects_by_name_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -455,7 +448,7 @@ def test_get_projects_by_name_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?name=My", + url=self.MYURL + "resource/api/projects?name=My&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -540,7 +533,7 @@ def test_get_projects_by_group(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -558,7 +551,7 @@ def test_get_projects_by_group_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -575,7 +568,7 @@ def test_get_projects_by_group_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -592,7 +585,7 @@ def test_get_projects_by_group_with_details(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?allDetails?group=SI", + url=self.MYURL + "resource/api/projects?allDetails=true&group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", # noqa: E501 body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -610,7 +603,7 @@ def test_get_projects_by_group_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?group=SI", + url=self.MYURL + "resource/api/projects?group=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{}', status=200, content_type="application/json", @@ -626,7 +619,7 @@ def test_get_projects_by_tag(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -644,7 +637,7 @@ def test_get_projects_by_tag_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{}', status=200, content_type="application/json", @@ -660,7 +653,7 @@ def test_get_projects_by_tag_invalid_reply(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -677,7 +670,7 @@ def test_get_projects_by_tag_invalid_reply2(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true", + url=self.MYURL + "resource/api/projects?tag=SI BP&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"xxsw360:projects": [{"name": "My Testproject", "externalIds": {"com.siemens.code.project.id": "13171"}}]}}', # noqa status=200, content_type="application/json", @@ -718,7 +711,7 @@ def test_get_project_vulnerabilities(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/vulnerabilities", + url=self.MYURL + "resource/api/projects/123/vulnerabilities?page=0&page_entries=50", body='{"_embedded": {"sw360:vulnerabilityDToes": [{"priority": "2 - major", "action": "Follow Recommendation"}]}}', # noqa status=200, content_type="application/json", @@ -741,7 +734,7 @@ def test_get_project_vulnerabilities_no_result(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/vulnerabilities", + url=self.MYURL + "resource/api/projects/123/vulnerabilities?&page=0&page_entries=50", body='{}', status=200, content_type="application/json", @@ -1011,7 +1004,7 @@ def test_update_project_releases_fresh_prj(self) -> None: lib = self.get_logged_in_lib() responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", json={}, ) responses.add( @@ -1023,7 +1016,7 @@ def test_update_project_releases_fresh_prj(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/124/releases?transitive=false", + url=self.MYURL + "resource/api/projects/124/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", json={'_embedded': {'sw360:projects': []}}, ) responses.add( @@ -1039,7 +1032,7 @@ def test_update_project_releases(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", body='{"_embedded": {"sw360:releases": [{"name": "ngx-device-detector ", "version": "1.3.20","_links": {"self": {"href": "https://sw360.siemens.com/resource/api/releases/3a4865e453873ee00d924469ff40f391" }}}]}}', # noqa status=200, content_type="application/json", @@ -1118,7 +1111,7 @@ def test_update_project_releases_failed(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects/123/releases?transitive=false", + url=self.MYURL + "resource/api/projects/123/releases?transitive=false&page=0&page_entries=50&sort=createdOn,desc", body='{"_embedded": {"sw360:releases": [{"name": "ngx-device-detector ", "version": "1.3.20","_links": {"self": {"href": "https://sw360.siemens.com/resource/api/releases/3a4865e453873ee00d924469ff40f391" }}}]}}', # noqa status=200, content_type="application/json",