From ec7def21efcecffb6d163e5604fec2c32e431724 Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Thu, 13 Aug 2026 23:34:55 +0530 Subject: [PATCH] feat(projects): update API calls to new v20 style 1. Update the functions exposed for Projects to use v20 style API calls. 2. Add pagination to calls where supported. 3. Add new utility for uploading attachments. Signed-off-by: Gaurav Mishra --- sw360/__init__.py | 8 +- sw360/base.py | 205 +++++++++++++++++++++++-- sw360/project.py | 283 ++++++++++++++++++++++++----------- sw360/sorting.py | 46 ++++++ sw360/sw360_api.py | 7 +- tests/test_sw360_projects.py | 77 +++++----- 6 files changed, 485 insertions(+), 141 deletions(-) create mode 100644 sw360/sorting.py 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",