From ec7def21efcecffb6d163e5604fec2c32e431724 Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Thu, 13 Aug 2026 23:34:55 +0530 Subject: [PATCH 1/9] 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", From c8c65f90ccafe13b01019a5d43a4847969922278 Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Sat, 15 Aug 2026 18:33:09 +0530 Subject: [PATCH 2/9] feat(components): update API calls to v20 style 1. Update the functions exposed for Components 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 diff --git c/sw360/components.py i/sw360/components.py index 6792d6e..59d945e 100644 --- c/sw360/components.py +++ i/sw360/components.py @@ -12,27 +12,70 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import ComponentSortColumn, SortParam from .sw360error import SW360Error class ComponentsMixin(BaseMixin): # return type List[Dict[str, Any]] | Optional[Dict[str, Any]] for Python 3.11 is good, # Union[List[Dict[str, Any]], Optional[Dict[str, Any]]] for lower Python versions is not good - def get_all_components(self, fields: str = "", page: int = -1, page_size: int = -1, - all_details: bool = False, - sort: str = "") -> Any: + + def __get_components_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: + """ + Take a pre-generated URL of components endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + components. + + :param url: Components 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 components (Sort by name if `None`) + :type sort: SortParam + :return: list of components + :rtype: list of JSON component 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 (resp and + "_embedded" in resp and + "sw360:components" in resp["_embedded"]): + return resp["_embedded"]["sw360:components"] + + return [] + + def get_all_components( + self, fields: str = "", page: int = -1, page_size: int = -1, + all_details: bool = False, sort: Optional[SortParam] = None + ) -> Any: """Get information of about all components API endpoint: GET /components + :param fields: Comma-separated fields in the components object to fetch + :type fields: string :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 all_details: retrieve all component details (optional)) + :param all_details: retrieve all component details (optional) :type all_details: bool - :param sort: sort order for the components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by name if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -47,48 +90,32 @@ class ComponentsMixin(BaseMixin): if fields: params["fields"] = fields - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort + if sort is None: + sort = ComponentSortColumn.NAME.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - if not resp: - return [] - - if "_embedded" not in resp: - return [] - - if "sw360:components" not in resp["_embedded"]: - return [] - - if page == -1: - return resp["_embedded"]["sw360:components"] - - return resp + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_components_by_type( - self, - component_type: str, - page: int = -1, - page_size: int = -1, - sort: str = "") -> List[Dict[str, Any]]: + self, component_type: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get information of about all components for certain type - API endpoint: GET /components + API endpoint: GET /components?type= :param component_type: the type of the component to be requested, one - of INTERNAL, OSS, COTS, FREESOFTWARE, INNER_SOURCE, SERVICE + of INTERNAL, OSS, COTS, FREESOFTWARE, INNER_SOURCE, SERVICE, + CODE_SNIPPET, COTS_TRUSTED_SUPPLIER :type component_type: string :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 components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by name if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -99,20 +126,13 @@ class ComponentsMixin(BaseMixin): fullbase_url = self.url + "resource/api/components" params = {"type": component_type} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort + if sort is None: + sort = ComponentSortColumn.NAME.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - - if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]): - return resp["_embedded"]["sw360:components"] - - return [] + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_component(self, component_id: str) -> Optional[Dict[str, Any]]: """Get information of about a component @@ -149,23 +169,21 @@ class ComponentsMixin(BaseMixin): return resp def get_component_by_name( - self, - component_name: str, - page: int = -1, - page_size: int = -1, - sort: str = "") -> Optional[Dict[str, Any]]: + self, component_name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Dict[str, Any]: """Get information of about a component - API endpoint: GET /components + API endpoint: GET /components?name= :param component_name: the name of the component to look for :type component_name: string :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 components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by score if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -176,16 +194,13 @@ class ComponentsMixin(BaseMixin): fullbase_url = self.url + "resource/api/components" params = {"name": component_name} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort + if sort is None: + sort = ComponentSortColumn.SCORE.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - return resp + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_components_by_external_id(self, ext_id_name: str, ext_id_value: str = "") -> List[Dict[str, Any]]: """Get components by external id. `ext_id_value` can be left blank to @@ -194,8 +209,8 @@ class ComponentsMixin(BaseMixin): API endpoint: GET /components :param ext_id_name: the name of the external id to look for - :param ext_id_value: the value of the external id to look for :type ext_id_name: string + :param ext_id_value: the value of the external id to look for :type ext_id_value: string :return: list of components :rtype: list of JSON component objects @@ -204,13 +219,12 @@ class ComponentsMixin(BaseMixin): if not ext_id_name: raise SW360Error(message="No external id name provided!") - resp = self.api_get( - self.url - + "resource/api/components/searchByExternalIds?" - + ext_id_name - + "=" - + ext_id_value - ) + fullbase_url = self.url + "resource/api/components/searchByExternalIds" + params = {ext_id_name: ext_id_value} + + url_with_param = self._add_params(fullbase_url, params) + + resp = self.api_get(url_with_param) if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]): return resp["_embedded"]["sw360:components"] @@ -225,7 +239,8 @@ class ComponentsMixin(BaseMixin): :param name: name of the new component :param description: description of the new component :param component_type: type of the new component, one of - "INTERNAL", "OSS", "COTS", "FREESOFTWARE", "INNER_SOURCE", "SERVICE", "CODE_SNIPPET" + "INTERNAL", "OSS", "COTS", "FREESOFTWARE", "INNER_SOURCE", "SERVICE", + "CODE_SNIPPET", "COTS_TRUSTED_SUPPLIER" :param homepage: homepage url of the new component :param component_details: further component details as defined by SW360 REST API :type name: string @@ -369,3 +384,33 @@ class ComponentsMixin(BaseMixin): return resp["_embedded"]["sw360:components"] return [] + + def upload_attachment_to_component( + self, component_id: str, upload_file: str, upload_type: str = "SOURCE", + upload_comment: str = "" + ) -> Optional[Dict[str, Any]]: + """Upload an attachment to a given Component. + + API endpoint: POST /attachments & PATCH /components/{id} + + :param component_id: the id of the Component + :type component_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 component id is missing or there is a negative HTTP response + """ + if not component_id: + raise SW360Error(message="No component id provided!") + + attachment_content = self._upload_resource_file(upload_file, upload_type, upload_comment) + attachment_content['attachmentType'] = upload_type # Make sure the type is correct + attachment_content['createdComment'] = upload_comment # Override + + current_component = self.get_component(component_id) + attachments = self._get_attachments(current_component) + attachments.append(attachment_content) + return self.update_component({'attachments': attachments}, component_id) diff --git c/sw360/sorting.py i/sw360/sorting.py index 0494661..224b821 100644 --- c/sw360/sorting.py +++ i/sw360/sorting.py @@ -44,3 +44,11 @@ class ReleaseSortColumn(BaseSortMixin, Enum): CLEARING_STATE = "clearingState" MAINLINE_STATE = "mainlineState" SCORE = "score" + +class ComponentSortColumn(BaseSortMixin, Enum): + SCORE = "score" + CREATED_ON = "createdOn" + NAME = "name" + VENDOR_NAMES = "vendorNames" + MAIN_LICENSE_IDS = "mainLicenseIds" + TYPE = "type" diff --git c/tests/test_sw360_components.py i/tests/test_sw360_components.py index 8d0d209..b9fde6c 100644 --- c/tests/test_sw360_components.py +++ i/tests/test_sw360_components.py @@ -18,6 +18,7 @@ import responses sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import ComponentSortColumn # noqa: E402 class Sw360TestComponents(unittest.TestCase): @@ -105,7 +106,7 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?fields=ownerCountry", # noqa + url=self.MYURL + "resource/api/components?fields=ownerCountry&luceneSearch=true&page=0&page_entries=50&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -127,15 +128,14 @@ class Sw360TestComponents(unittest.TestCase): self.assertTrue(actual) responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?fields=ownerCountry&page=1&page_entries=2", # noqa + url=self.MYURL + "resource/api/components?fields=ownerCountry&page=1&page_entries=2&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - data = lib.get_all_components("ownerCountry", 1, 2) - components = data["_embedded"]["sw360:components"] + components = lib.get_all_components("ownerCountry", 1, 2) self.assertIsNotNone(components) self.assertTrue(len(components) > 0) self.assertEqual("Tethys.Logging", components[0]["name"]) @@ -151,7 +151,7 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?allDetails=true", # noqa + url=self.MYURL + "resource/api/components?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -174,14 +174,14 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?allDetails=true&sort=name%2Cdesc", # noqa + url=self.MYURL + "resource/api/components?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,desc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) > 0) self.assertEqual("Tethys.Logging", components[0]["name"]) @@ -204,7 +204,7 @@ class Sw360TestComponents(unittest.TestCase): adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) == 0) @@ -225,7 +225,7 @@ class Sw360TestComponents(unittest.TestCase): adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) == 0) @@ -239,7 +239,7 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -262,7 +262,7 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{}', status=200, content_type="application/json", @@ -282,7 +282,7 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_xxembedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -303,7 +303,7 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"xxsw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -366,7 +366,7 @@ class Sw360TestComponents(unittest.TestCase): responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?name=MyComponent", + url=self.MYURL + "resource/api/components?name=MyComponent&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"name": "MyComponent"}', status=200, content_type="application/json", --- sw360/components.py | 195 ++++++++++++++++++++------------- sw360/sorting.py | 8 ++ tests/test_sw360_components.py | 28 ++--- 3 files changed, 142 insertions(+), 89 deletions(-) diff --git a/sw360/components.py b/sw360/components.py index 6792d6e..59d945e 100644 --- a/sw360/components.py +++ b/sw360/components.py @@ -12,27 +12,70 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import ComponentSortColumn, SortParam from .sw360error import SW360Error class ComponentsMixin(BaseMixin): # return type List[Dict[str, Any]] | Optional[Dict[str, Any]] for Python 3.11 is good, # Union[List[Dict[str, Any]], Optional[Dict[str, Any]]] for lower Python versions is not good - def get_all_components(self, fields: str = "", page: int = -1, page_size: int = -1, - all_details: bool = False, - sort: str = "") -> Any: + + def __get_components_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: + """ + Take a pre-generated URL of components endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + components. + + :param url: Components 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 components (Sort by name if `None`) + :type sort: SortParam + :return: list of components + :rtype: list of JSON component 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 (resp and + "_embedded" in resp and + "sw360:components" in resp["_embedded"]): + return resp["_embedded"]["sw360:components"] + + return [] + + def get_all_components( + self, fields: str = "", page: int = -1, page_size: int = -1, + all_details: bool = False, sort: Optional[SortParam] = None + ) -> Any: """Get information of about all components API endpoint: GET /components + :param fields: Comma-separated fields in the components object to fetch + :type fields: string :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 all_details: retrieve all component details (optional)) + :param all_details: retrieve all component details (optional) :type all_details: bool - :param sort: sort order for the components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by name if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -47,48 +90,32 @@ def get_all_components(self, fields: str = "", page: int = -1, page_size: int = if fields: params["fields"] = fields - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - if not resp: - return [] - - if "_embedded" not in resp: - return [] - - if "sw360:components" not in resp["_embedded"]: - return [] - - if page == -1: - return resp["_embedded"]["sw360:components"] + if sort is None: + sort = ComponentSortColumn.NAME.asc() - return resp + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_components_by_type( - self, - component_type: str, - page: int = -1, - page_size: int = -1, - sort: str = "") -> List[Dict[str, Any]]: + self, component_type: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Get information of about all components for certain type - API endpoint: GET /components + API endpoint: GET /components?type= :param component_type: the type of the component to be requested, one - of INTERNAL, OSS, COTS, FREESOFTWARE, INNER_SOURCE, SERVICE + of INTERNAL, OSS, COTS, FREESOFTWARE, INNER_SOURCE, SERVICE, + CODE_SNIPPET, COTS_TRUSTED_SUPPLIER :type component_type: string :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 components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by name if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -99,20 +126,13 @@ def get_components_by_type( fullbase_url = self.url + "resource/api/components" params = {"type": component_type} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + url_with_param = self._add_params(fullbase_url, params) - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + if sort is None: + sort = ComponentSortColumn.NAME.asc() - if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]): - return resp["_embedded"]["sw360:components"] - - return [] + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_component(self, component_id: str) -> Optional[Dict[str, Any]]: """Get information of about a component @@ -149,23 +169,21 @@ def get_component_by_url(self, component_url: str) -> Optional[Dict[str, Any]]: return resp def get_component_by_name( - self, - component_name: str, - page: int = -1, - page_size: int = -1, - sort: str = "") -> Optional[Dict[str, Any]]: + self, component_name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Dict[str, Any]: """Get information of about a component - API endpoint: GET /components + API endpoint: GET /components?name= :param component_name: the name of the component to look for :type component_name: string :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 components ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by score if `None`) + :type sort: SortParam :return: list of components :rtype: list of JSON component objects :raises SW360Error: if there is a negative HTTP response @@ -176,16 +194,13 @@ def get_component_by_name( fullbase_url = self.url + "resource/api/components" params = {"name": component_name} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort + if sort is None: + sort = ComponentSortColumn.SCORE.asc() - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - return resp + return self.__get_components_filtered(url_with_param, page, page_size, + sort) def get_components_by_external_id(self, ext_id_name: str, ext_id_value: str = "") -> List[Dict[str, Any]]: """Get components by external id. `ext_id_value` can be left blank to @@ -194,8 +209,8 @@ def get_components_by_external_id(self, ext_id_name: str, ext_id_value: str = "" API endpoint: GET /components :param ext_id_name: the name of the external id to look for - :param ext_id_value: the value of the external id to look for :type ext_id_name: string + :param ext_id_value: the value of the external id to look for :type ext_id_value: string :return: list of components :rtype: list of JSON component objects @@ -204,13 +219,12 @@ def get_components_by_external_id(self, ext_id_name: str, ext_id_value: str = "" if not ext_id_name: raise SW360Error(message="No external id name provided!") - resp = self.api_get( - self.url - + "resource/api/components/searchByExternalIds?" - + ext_id_name - + "=" - + ext_id_value - ) + fullbase_url = self.url + "resource/api/components/searchByExternalIds" + params = {ext_id_name: ext_id_value} + + url_with_param = self._add_params(fullbase_url, params) + + resp = self.api_get(url_with_param) if resp and ("_embedded" in resp) and ("sw360:components" in resp["_embedded"]): return resp["_embedded"]["sw360:components"] @@ -225,7 +239,8 @@ def create_new_component(self, name: str, description: str, component_type: str, :param name: name of the new component :param description: description of the new component :param component_type: type of the new component, one of - "INTERNAL", "OSS", "COTS", "FREESOFTWARE", "INNER_SOURCE", "SERVICE", "CODE_SNIPPET" + "INTERNAL", "OSS", "COTS", "FREESOFTWARE", "INNER_SOURCE", "SERVICE", + "CODE_SNIPPET", "COTS_TRUSTED_SUPPLIER" :param homepage: homepage url of the new component :param component_details: further component details as defined by SW360 REST API :type name: string @@ -369,3 +384,33 @@ def get_recent_components(self) -> Optional[List[Dict[str, Any]]]: return resp["_embedded"]["sw360:components"] return [] + + def upload_attachment_to_component( + self, component_id: str, upload_file: str, upload_type: str = "SOURCE", + upload_comment: str = "" + ) -> Optional[Dict[str, Any]]: + """Upload an attachment to a given Component. + + API endpoint: POST /attachments & PATCH /components/{id} + + :param component_id: the id of the Component + :type component_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 component id is missing or there is a negative HTTP response + """ + if not component_id: + raise SW360Error(message="No component id provided!") + + attachment_content = self._upload_resource_file(upload_file, upload_type, upload_comment) + attachment_content['attachmentType'] = upload_type # Make sure the type is correct + attachment_content['createdComment'] = upload_comment # Override + + current_component = self.get_component(component_id) + attachments = self._get_attachments(current_component) + attachments.append(attachment_content) + return self.update_component({'attachments': attachments}, component_id) diff --git a/sw360/sorting.py b/sw360/sorting.py index 0494661..224b821 100644 --- a/sw360/sorting.py +++ b/sw360/sorting.py @@ -44,3 +44,11 @@ class ReleaseSortColumn(BaseSortMixin, Enum): CLEARING_STATE = "clearingState" MAINLINE_STATE = "mainlineState" SCORE = "score" + +class ComponentSortColumn(BaseSortMixin, Enum): + SCORE = "score" + CREATED_ON = "createdOn" + NAME = "name" + VENDOR_NAMES = "vendorNames" + MAIN_LICENSE_IDS = "mainLicenseIds" + TYPE = "type" diff --git a/tests/test_sw360_components.py b/tests/test_sw360_components.py index 8d0d209..b9fde6c 100644 --- a/tests/test_sw360_components.py +++ b/tests/test_sw360_components.py @@ -18,6 +18,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import ComponentSortColumn # noqa: E402 class Sw360TestComponents(unittest.TestCase): @@ -105,7 +106,7 @@ def test_get_all_components_with_fields(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?fields=ownerCountry", # noqa + url=self.MYURL + "resource/api/components?fields=ownerCountry&luceneSearch=true&page=0&page_entries=50&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -127,15 +128,14 @@ def test_get_all_components_with_fields_and_paging(self) -> None: self.assertTrue(actual) responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?fields=ownerCountry&page=1&page_entries=2", # noqa + url=self.MYURL + "resource/api/components?fields=ownerCountry&page=1&page_entries=2&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - data = lib.get_all_components("ownerCountry", 1, 2) - components = data["_embedded"]["sw360:components"] + components = lib.get_all_components("ownerCountry", 1, 2) self.assertIsNotNone(components) self.assertTrue(len(components) > 0) self.assertEqual("Tethys.Logging", components[0]["name"]) @@ -151,7 +151,7 @@ def test_get_all_components_with_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?allDetails=true", # noqa + url=self.MYURL + "resource/api/components?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -174,14 +174,14 @@ def test_get_all_components_with_all_details_and_sorting(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?allDetails=true&sort=name%2Cdesc", # noqa + url=self.MYURL + "resource/api/components?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,desc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) > 0) self.assertEqual("Tethys.Logging", components[0]["name"]) @@ -204,7 +204,7 @@ def test_get_all_components_invalid_reply(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) == 0) @@ -225,7 +225,7 @@ def test_get_all_components_invalid_reply2(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - components = lib.get_all_components(all_details=True, sort="name,desc") + components = lib.get_all_components(all_details=True, sort=ComponentSortColumn.NAME.desc()) self.assertIsNotNone(components) self.assertTrue(len(components) == 0) @@ -239,7 +239,7 @@ def test_get_all_components_by_type(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -262,7 +262,7 @@ def test_get_all_components_by_type_no_result(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{}', status=200, content_type="application/json", @@ -282,7 +282,7 @@ def test_get_all_components_by_type_invalid_reply(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_xxembedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -303,7 +303,7 @@ def test_get_all_components_by_type_invalid_reply2(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?type=OSS", + url=self.MYURL + "resource/api/components?type=OSS&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"xxsw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", @@ -366,7 +366,7 @@ def test_get_component_by_name(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?name=MyComponent", + url=self.MYURL + "resource/api/components?name=MyComponent&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"name": "MyComponent"}', status=200, content_type="application/json", From 1bfe8f794844fec6db8d0c920b24df8e44ffd298 Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Sun, 16 Aug 2026 21:10:13 +0530 Subject: [PATCH 3/9] feat(license): update API calls to new v20 style Signed-off-by: Gaurav Mishra --- sw360/clearing.py | 4 ++-- sw360/license.py | 47 ++++++++++++++++++++++++++++++++++++----------- sw360/sorting.py | 5 +++++ 3 files changed, 43 insertions(+), 13 deletions(-) diff --git a/sw360/clearing.py b/sw360/clearing.py index c2892f9..f4d2549 100644 --- a/sw360/clearing.py +++ b/sw360/clearing.py @@ -39,8 +39,8 @@ def get_clearing_request_for_project(self, project_id: str) -> Optional[Dict[str API endpoint: GET /clearingrequest/project/{id} - :param request_id: the id of the clearing request to be requested - :type request_id: string + :param project_id: the id of the Project to fetch clearing requests for + :type project_id: string :return: a clearing request :rtype: JSON clearing request object :raises SW360Error: if there is a negative HTTP response diff --git a/sw360/license.py b/sw360/license.py index f54143c..12cac92 100644 --- a/sw360/license.py +++ b/sw360/license.py @@ -14,6 +14,7 @@ import requests from .base import BaseMixin +from .sorting import LicenseSortColumn from .sw360error import SW360Error @@ -24,7 +25,6 @@ def create_new_license( fullName: str, text: str, checked: bool = False, - license_details: Dict[str, Any] = {}, ) -> Any: """Create a new license @@ -34,8 +34,8 @@ def create_new_license( :param fullName: descriptive license name :param text: license text :param checked: if license is checked - :type shortname: string - :type fullname: string + :type shortName: string + :type fullName: string :type text: string :type checked: bool :return: SW360 result @@ -53,10 +53,12 @@ def create_new_license( url = self.url + "resource/api/licenses" - license_details["shortName"] = shortName - license_details["fullName"] = fullName - license_details["text"] = text - license_details["checked"] = checked + license_details = { + "shortName": shortName, + "fullName": fullName, + "text": text, + "checked": checked, + } response = self.api_post(url, json=license_details) if response is not None: @@ -67,7 +69,7 @@ def create_new_license( def delete_license(self, license_shortname: str) -> Optional[bool]: """Delete an existing license - API endpoint: PATCH /licenses + API endpoint: DELETE /licenses :param license_shortname: license shortname as the id :type license_shortname: string @@ -76,8 +78,6 @@ def delete_license(self, license_shortname: str) -> Optional[bool]: :raises SW360Error: if there is a negative HTTP response """ - # 2025-01-23: returns 500 - internal server error - if not license_shortname: raise SW360Error(message="No license shortname provided!") @@ -132,7 +132,9 @@ def get_all_licenses(self) -> List[Dict[str, Any]]: :raises SW360Error: if there is a negative HTTP response """ - resp = self.api_get(self.url + "resource/api/licenses") + fullbase_url = self.url + "resource/api/licenses" + resp = self.api_get_all(fullbase_url) + if resp and "_embedded" in resp and "sw360:licenses" in resp["_embedded"]: return resp["_embedded"]["sw360:licenses"] @@ -154,3 +156,26 @@ def get_license(self, license_id: str) -> Optional[Dict[str, Any]]: resp = self.api_get(self.url + "resource/api/licenses/" + license_id) return resp + + def search_license(self, search_text: str) -> Optional[Dict[str, Any]]: + """Get search a license by fullName, shortName or text + + API endpoint: GET /licenses?searchText= + + :param search_text: the search string + :type search_text: string + :return: list of licenses + :rtype: list of JSON license objects + :raises SW360Error: if there is a negative HTTP response + """ + if not search_text: + raise SW360Error(message="Search Text provided!") + + fullbase_url = self.url + "resource/api/licenses" + params = {"searchText": search_text} + full_url = self._add_params(fullbase_url, params) + + sort = LicenseSortColumn.SCORE.asc() + + resp = self.api_get_all(full_url, sort) + return resp diff --git a/sw360/sorting.py b/sw360/sorting.py index 224b821..fc68177 100644 --- a/sw360/sorting.py +++ b/sw360/sorting.py @@ -52,3 +52,8 @@ class ComponentSortColumn(BaseSortMixin, Enum): VENDOR_NAMES = "vendorNames" MAIN_LICENSE_IDS = "mainLicenseIds" TYPE = "type" + +class LicenseSortColumn(BaseSortMixin, Enum): + SCORE = "score" + FULL_NAME = "fullName" + SHORT_NAME = "shortName" From 87161ad11d434f508dc2aef8ab057ffdb77f18a6 Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Sun, 16 Aug 2026 21:16:44 +0530 Subject: [PATCH 4/9] feat(moderation): update API calls to v20 style Signed-off-by: Gaurav Mishra --- sw360/moderationrequests.py | 68 ++++++++++++++++---------- sw360/sorting.py | 10 ++++ tests/test_sw360_moderationrequests.py | 9 ++-- 3 files changed, 56 insertions(+), 31 deletions(-) diff --git a/sw360/moderationrequests.py b/sw360/moderationrequests.py index 4027d9d..79c2aa7 100644 --- a/sw360/moderationrequests.py +++ b/sw360/moderationrequests.py @@ -12,55 +12,66 @@ from sw360.sw360error import SW360Error from .base import BaseMixin +from .sorting import ModerationSortColumn, SortParam class ModerationRequestMixin(BaseMixin): - def get_all_moderation_requests(self, page: int = -1, page_size: int = -1, - sort: str = "") -> Optional[Dict[str, Any]]: + def get_all_moderation_requests( + self, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Optional[Dict[str, Any]]: """Get information of about all moderation requests API endpoint: GET /moderationrequest :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 packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by request date if + `None`) + :type sort: SortParam :return: list of moderation requests :rtype: list of JSON moderation requests objects :raises SW360Error: if there is a negative HTTP response """ fullbase_url = self.url + "resource/api/moderationrequest" - params = {} + params = {"luceneSearch": "true"} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + if sort is None: + sort = ModerationSortColumn.REQUEST_DATE.desc() full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + return resp - def get_moderation_requests_by_state(self, state: str, all_details: bool = False, - page: int = -1, page_size: int = -1, - sort: str = "") -> Optional[Dict[str, Any]]: + def get_moderation_requests_by_state( + self, state: str, all_details: bool = False, page: int = -1, + page_size: int = -1, sort: Optional[SortParam] = None + ) -> Optional[Dict[str, Any]]: """Get information of about all moderation requests API endpoint: GET /moderationrequest/byState - :param all_details: retrieve all package details (optional)) + :param state: State of the moderation request. One of "open", "closed" + :type state: string + :param all_details: retrieve all moderation 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 packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by request date if + `None`) + :type sort: SortParam :return: list of moderation requests :rtype: list of JSON moderation requests objects :raises SW360Error: if there is a negative HTTP response @@ -74,15 +85,18 @@ def get_moderation_requests_by_state(self, state: str, all_details: bool = False if all_details: params["allDetails"] = "true" - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort + if sort is None: + sort = ModerationSortColumn.REQUEST_DATE.desc() full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) + return resp def get_moderation_request(self, mr_id: str) -> Optional[Dict[str, Any]]: diff --git a/sw360/sorting.py b/sw360/sorting.py index fc68177..a3383d0 100644 --- a/sw360/sorting.py +++ b/sw360/sorting.py @@ -57,3 +57,13 @@ class LicenseSortColumn(BaseSortMixin, Enum): SCORE = "score" FULL_NAME = "fullName" SHORT_NAME = "shortName" + +class ModerationSortColumn(BaseSortMixin, Enum): + SCORE = "score" + DOCUMENT_NAME = "documentName" + DOCUMENT_TYPE = "documentType" + COMPONENT_TYPE = "componentType" + MODERATION_STATE = "moderationState" + REQUEST_DATE = "requestDate" + REQUESTING_USER = "requestingUser" + REQUESTING_USER_DEPARTMENT = "requestingUserDepartment" diff --git a/tests/test_sw360_moderationrequests.py b/tests/test_sw360_moderationrequests.py index b53f472..44b4cbd 100644 --- a/tests/test_sw360_moderationrequests.py +++ b/tests/test_sw360_moderationrequests.py @@ -14,6 +14,7 @@ import responses from sw360 import SW360 +from sw360.sorting import ModerationSortColumn sys.path.insert(1, "..") @@ -50,7 +51,7 @@ def test_get_all_moderation_requests(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/moderationrequest?page=2&page_entries=8&sort=timestamp%2Cdesc", + url=self.MYURL + "resource/api/moderationrequest?luceneSearch=true&page=2&page_entries=8&sort=requestDate%2Cdesc", body='''{ "_embedded": { "sw360:moderationRequests": [ @@ -104,7 +105,7 @@ def test_get_all_moderation_requests(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - mrs = lib.get_all_moderation_requests(page=2, page_size=8, sort="timestamp,desc") + mrs = lib.get_all_moderation_requests(page=2, page_size=8, sort=ModerationSortColumn.REQUEST_DATE.desc()) self.assertIsNotNone(mrs) if mrs: self.assertTrue(len(mrs) > 0) @@ -120,7 +121,7 @@ def test_get_moderation_requests_by_state(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/moderationrequest/byState?state=open&allDetails=true&page=2&page_entries=8&sort=timestamp%2Cdesc", # noqa + url=self.MYURL + "resource/api/moderationrequest/byState?state=open&allDetails=true&page=2&page_entries=8&sort=requestDate%2Cdesc", # noqa body='''{ "_embedded": { "sw360:moderationRequests": [ @@ -174,7 +175,7 @@ def test_get_moderation_requests_by_state(self) -> None: adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - mrs = lib.get_moderation_requests_by_state("open", True, page=2, page_size=8, sort="timestamp,desc") + mrs = lib.get_moderation_requests_by_state("open", True, page=2, page_size=8, sort=ModerationSortColumn.REQUEST_DATE.desc()) self.assertIsNotNone(mrs) if mrs: self.assertTrue(len(mrs) > 0) From 6864478331e1c764fca1ae77e19f9d1d2c4b8140 Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Mon, 17 Aug 2026 00:35:10 +0530 Subject: [PATCH 5/9] fix(api): URL creation in project and component Signed-off-by: Gaurav Mishra --- sw360/components.py | 2 +- sw360/project.py | 2 +- tests/test_sw360_components.py | 2 +- tests/test_sw360_projects.py | 4 ++-- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/sw360/components.py b/sw360/components.py index 59d945e..794faa4 100644 --- a/sw360/components.py +++ b/sw360/components.py @@ -44,7 +44,7 @@ def __get_components_filtered( 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) + full_url = self._add_pagination(full_url, page, page_size, sort) if page_size == -1: resp = self.api_get_all(full_url, sort) diff --git a/sw360/project.py b/sw360/project.py index 2d5e40e..130c77c 100644 --- a/sw360/project.py +++ b/sw360/project.py @@ -99,7 +99,7 @@ def __get_projects_filtered( 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) + full_url = self._add_pagination(full_url, page, page_size, sort) if page_size == -1: resp = self.api_get_all(full_url, sort) diff --git a/tests/test_sw360_components.py b/tests/test_sw360_components.py index b9fde6c..029a1cf 100644 --- a/tests/test_sw360_components.py +++ b/tests/test_sw360_components.py @@ -128,7 +128,7 @@ def test_get_all_components_with_fields_and_paging(self) -> None: self.assertTrue(actual) responses.add( method=responses.GET, - url=self.MYURL + "resource/api/components?fields=ownerCountry&page=1&page_entries=2&sort=name,asc", # noqa + url=self.MYURL + "resource/api/components?fields=ownerCountry&luceneSearch=true&page=1&page_entries=2&sort=name,asc", # noqa body='{"_embedded": {"sw360:components": [{"name": "Tethys.Logging", "ownerCountry": "DE", "componentType": "OSS", "externalIds": {"package-url": "pkg:nuget/Tethys.Logging"}}]}}', # noqa status=200, content_type="application/json", diff --git a/tests/test_sw360_projects.py b/tests/test_sw360_projects.py index 38b4d53..7d4e3ca 100644 --- a/tests/test_sw360_projects.py +++ b/tests/test_sw360_projects.py @@ -222,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&sort=name,asc", + url=self.MYURL + "resource/api/projects?luceneSearch=true&page=1&page_entries=2&sort=name,asc", body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", @@ -240,7 +240,7 @@ def test_get_projects_with_paging_and_details(self) -> None: responses.add( responses.GET, - url=self.MYURL + "resource/api/projects?allDetails=true&page=3&page_entries=4&sort=name%2Cdesc", # noqa + url=self.MYURL + "resource/api/projects?allDetails=true&luceneSearch=true&page=3&page_entries=4&sort=name%2Cdesc", # noqa body='{"_embedded": {"sw360:projects": [{"name": "My Testproject"}]}}', status=200, content_type="application/json", From fb2c6f50fa4ac988fced8cb9ee48eb23e407f77a Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Sun, 16 Aug 2026 21:17:26 +0530 Subject: [PATCH 6/9] feat(packages): update API calls to v20 style Signed-off-by: Gaurav Mishra --- sw360/packages.py | 167 +++++++++++++++++++++++++---------- sw360/sorting.py | 7 ++ tests/test_sw360_packages.py | 25 +++--- 3 files changed, 139 insertions(+), 60 deletions(-) diff --git a/sw360/packages.py b/sw360/packages.py index e86d24f..02f7eee 100644 --- a/sw360/packages.py +++ b/sw360/packages.py @@ -10,10 +10,50 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import PackageSortColumn, SortParam from .sw360error import SW360Error class PackagesMixin(BaseMixin): + + def __get_packages_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: + """ + Take a pre-generated URL of packages endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + packages. + + :param url: Packages 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 package + :type sort: SortParam + :return: list of packages + :rtype: list of JSON package 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(full_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 (resp and + "_embedded" in resp and + "sw360:packages" in resp["_embedded"]): + return resp["_embedded"]["sw360:packages"] + + return [] + def get_package(self, package_id: str) -> Optional[Dict[str, Any]]: """Get information of about a package @@ -31,13 +71,22 @@ def get_package(self, package_id: str) -> Optional[Dict[str, Any]]: resp = self.api_get(self.url + "resource/api/packages/" + package_id) return resp - def get_packages_by_name(self, name: str) -> List[Any]: + def get_packages_by_name( + self, name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Any]: """Gets a list of packages that match the given name. API endpoint: GET /packages?name= :param name: the name :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 packages (Sort by score if `None`) + :type sort: SortParam :return: list of packages :rtype: list of JSON package objects :raises SW360Error: if there is a negative HTTP response @@ -45,28 +94,44 @@ def get_packages_by_name(self, name: str) -> List[Any]: if not name: raise SW360Error(message="No package name provided!") - full_url = self.url + "resource/api/packages?name=" + name - resp = self.api_get(full_url) - if resp and ("_embedded" in resp) and ("sw360:packages" in resp["_embedded"]): - return resp["_embedded"]["sw360:packages"] + fullbase_url = self.url + "resource/api/packages" + params = {"name": name} - return [] + url_with_param = self._add_params(fullbase_url, params) + + if sort is None: + sort = PackageSortColumn.SCORE.asc() - def get_all_packages(self, name: str = "", version: str = "", purl: str = "", - all_details: bool = False, page: int = -1, - page_size: int = -1, sort: str = "") -> Any: + return self.__get_packages_filtered(url_with_param, page, page_size, + sort) + + def get_all_packages( + self, name: str = "", version: str = "", purl: str = "", + all_details: bool = False, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: """Get information of about all packages API endpoint: GET /releases - :param all_details: retrieve all package details (optional)) + :param name: name to filter + :type name: str + :param version: version to filter + :type version: str + :param purl: purl to filter + :type purl: str + :param all_details: retrieve all package 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 packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (name default, score if + filtering) + :type sort: SortParam + :return: list of packages + :rtype: list of JSON package objects + :raises SW360Error: if there is a negative HTTP response :return: list of packages :rtype: list of JSON package objects :raises SW360Error: if there is a negative HTTP response @@ -86,35 +151,32 @@ def get_all_packages(self, name: str = "", version: str = "", purl: str = "", if purl: params["purl"] = purl - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) + url_with_param = self._add_params(fullbase_url, params) - if page == -1 and resp and ("_embedded" in resp) and ("sw360:packages" in resp["_embedded"]): - return resp["_embedded"]["sw360:packages"] + if sort is None: + sort = PackageSortColumn.NAME.asc() + if name != "" or version != "" or purl != "": + sort = PackageSortColumn.SCORE.asc() - return resp + return self.__get_packages_filtered(url_with_param, page, page_size, + sort) - def get_packages_by_packagemanager(self, manager: str, page: int = -1, - page_size: int = -1, sort: str = "") -> Any: + def get_packages_by_packagemanager( + self, manager: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: """Get information of about all packages of a specific package manager - API endpoint: GET /releases + API endpoint: GET /releases?packageManager= :param manager: name of the package manager :type manager: str :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 packages ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the components (Sort by score if `None`) + :type sort: SortParam :return: list of packages :rtype: list of JSON package objects :raises SW360Error: if there is a negative HTTP response @@ -125,20 +187,13 @@ def get_packages_by_packagemanager(self, manager: str, page: int = -1, fullbase_url = self.url + "resource/api/packages" params = {"packageManager": manager} - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) + url_with_param = self._add_params(fullbase_url, params) - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - resp = self.api_get(full_url) - - if page == -1 and resp and ("_embedded" in resp) and ("sw360:packages" in resp["_embedded"]): - return resp["_embedded"]["sw360:packages"] + if sort is None: + sort = PackageSortColumn.SCORE.asc() - return resp + return self.__get_packages_filtered(url_with_param, page, page_size, + sort) def create_new_package(self, name: str, version: str, purl: str, package_type: str, package_details: Dict[str, Any] = {}) -> Optional[Dict[str, Any]]: @@ -185,10 +240,10 @@ def create_new_package(self, name: str, version: str, purl: str, def update_package(self, package: Dict[str, Any], package_id: str) -> Optional[Dict[str, Any]]: """Update an existing package - API endpoint: PATCH /packages + API endpoint: PATCH /packages/{id} - :param release: the new package data - :param release_id: the id of the package to be updated + :param package: the new package data + :param package_id: the id of the package to be updated :type package: JSON :type package_id: string :return: SW360 result @@ -205,7 +260,7 @@ def update_package(self, package: Dict[str, Any], package_id: str) -> Optional[D def delete_package(self, package_id: str) -> Optional[Dict[str, Any]]: """Delete an existing package - API endpoint: DELETE /packages + API endpoint: DELETE /packages/{id} :param package_id: the id of the package to be deleted :type package_id: string @@ -224,3 +279,21 @@ def delete_package(self, package_id: str) -> Optional[Dict[str, Any]]: if response.text: return response.json() return None + + def get_users_of_package(self, package_id: str) -> Optional[Dict[str, Any]]: + """Get information of about the uses of a package + + API endpoint: GET /packages/{id}/usage + + :param package_id: the id of the package to be requested + :type package_id: string + :return: all uses of this package + :rtype: JSON objects + :raises SW360Error: if there is a negative HTTP response + """ + if not package_id: + raise SW360Error(message="No package id provided!") + + resp = self.api_get(self.url + "resource/api/packages/"+ package_id + + "/usage") + return resp diff --git a/sw360/sorting.py b/sw360/sorting.py index a3383d0..5a73092 100644 --- a/sw360/sorting.py +++ b/sw360/sorting.py @@ -67,3 +67,10 @@ class ModerationSortColumn(BaseSortMixin, Enum): REQUEST_DATE = "requestDate" REQUESTING_USER = "requestingUser" REQUESTING_USER_DEPARTMENT = "requestingUserDepartment" + +class PackageSortColumn(BaseSortMixin, Enum): + SCORE = "score" + CREATED_ON = "createdOn" + NAME = "name" + VERSION = "version" + PACKAGE_MANAGER = "packageManager" diff --git a/tests/test_sw360_packages.py b/tests/test_sw360_packages.py index 4c067b3..f692a72 100644 --- a/tests/test_sw360_packages.py +++ b/tests/test_sw360_packages.py @@ -17,6 +17,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import PackageSortColumn # noqa: E402 class Sw360TestPackages(unittest.TestCase): @@ -106,7 +107,7 @@ def test_get_all_packages_with_fields_and_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?allDetails=true&purl=pkg:pypi/cli-support@2.0.0", + url=self.MYURL + "resource/api/packages?allDetails=true&purl=pkg:pypi/cli-support@2.0.0&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:packages": [{"name": "Tethys.Logging", "version": "1.3.0", "packageType": "FRAMEWORK"}]}}', # noqa status=200, content_type="application/json", @@ -128,19 +129,18 @@ def test_get_all_packages_with_fields_and_all_details2(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?allDetails=true&name=cli-support&version=2.0.0&page=2&page_entries=6&sort=name%2Cdesc", # noqa + url=self.MYURL + "resource/api/packages?allDetails=true&name=cli-support&version=2.0.0&luceneSearch=true&page=2&page_entries=6&sort=name,desc", # noqa body='{"_embedded": {"sw360:packages": [{"name": "Tethys.Logging", "version": "1.3.0", "packageType": "FRAMEWORK"}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - packages = lib.get_all_packages(name="cli-support", version="2.0.0", all_details=True, page=2, page_size=6, sort="name,desc") # noqa + packages = lib.get_all_packages(name="cli-support", version="2.0.0", all_details=True, page=2, page_size=6, sort=PackageSortColumn.NAME.desc()) # noqa self.assertIsNotNone(packages) self.assertTrue(len(packages) > 0) - pkgs = packages["_embedded"]["sw360:packages"] - self.assertEqual("Tethys.Logging", pkgs[0]["name"]) - self.assertEqual("1.3.0", pkgs[0]["version"]) + self.assertEqual("Tethys.Logging", packages[0]["name"]) + self.assertEqual("1.3.0", packages[0]["version"]) @responses.activate def test_get_packages_by_name(self) -> None: @@ -151,7 +151,7 @@ def test_get_packages_by_name(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?name=john", + url=self.MYURL + "resource/api/packages?name=john&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:packages": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/packages/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", @@ -173,7 +173,7 @@ def test_get_packages_by_packagemanager(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?packageManager=nuget", + url=self.MYURL + "resource/api/packages?packageManager=nuget&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:packages": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/packages/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", @@ -195,19 +195,18 @@ def test_get_packages_by_packagemanager_with_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/packages?packageManager=nuget&page=1&page_entries=5&sort=name%2Cdesc", + url=self.MYURL + "resource/api/packages?packageManager=nuget&luceneSearch=true&page=1&page_entries=5&sort=name%2Cdesc", body='{"_embedded": {"sw360:packages": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/packages/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - packages = lib.get_packages_by_packagemanager("nuget", page=1, page_size=5, sort="name,desc") + packages = lib.get_packages_by_packagemanager("nuget", page=1, page_size=5, sort=PackageSortColumn.NAME.desc()) self.assertIsNotNone(packages) self.assertTrue(len(packages) > 0) - pkgs = packages["_embedded"]["sw360:packages"] - self.assertEqual("john", pkgs[0]["name"]) - self.assertEqual("2.2.2", pkgs[0]["version"]) + self.assertEqual("john", packages[0]["name"]) + self.assertEqual("2.2.2", packages[0]["version"]) @responses.activate def test_create_new_package(self) -> None: From 2d19253623c094601be3c9a26879084cd200a661 Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Sun, 16 Aug 2026 21:17:54 +0530 Subject: [PATCH 7/9] feat(releases): update API calls to v20 style Signed-off-by: Gaurav Mishra --- sw360/releases.py | 154 +++++++++++++++++++++++++++-------- tests/test_sw360_releases.py | 22 ++--- 2 files changed, 129 insertions(+), 47 deletions(-) diff --git a/sw360/releases.py b/sw360/releases.py index 3ad5371..5bed6cd 100644 --- a/sw360/releases.py +++ b/sw360/releases.py @@ -12,10 +12,50 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import ReleaseSortColumn, SortParam from .sw360error import SW360Error class ReleasesMixin(BaseMixin): + + def __get_releases_filtered( + self, url: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> Any: + """ + Take a pre-generated URL of releases endpoint, with filters applied. + Then call the API with appropriate pagination and sorting to get the + releases. + + :param url: Releases 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 releases + :type sort: SortParam + :return: list of releases + :rtype: list of JSON release 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(full_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 (resp and + "_embedded" in resp and + "sw360:releases" in resp["_embedded"]): + return resp["_embedded"]["sw360:releases"] + + return [] + def get_release(self, release_id: str) -> Optional[Dict[str, Any]]: """Get information of about a release @@ -38,8 +78,8 @@ def get_release_by_url(self, release_url: str) -> Optional[Dict[str, Any]]: API endpoint: GET /releases - :param url: the full url of the release to be requested - :type url: string + :param release_url: the full url of the release to be requested + :type release_url: string :return: a release :rtype: JSON release object :raises SW360Error: if there is a negative HTTP response @@ -50,13 +90,22 @@ def get_release_by_url(self, release_url: str) -> Optional[Dict[str, Any]]: resp = self.api_get(release_url) return resp - def get_releases_by_name(self, name: str) -> List[Any]: + def get_releases_by_name( + self, name: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Any]: """Gets a list of releases that match the given name. API endpoint: GET /releases?name= :param name: the name :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 releases (Sort by score if `None`) + :type sort: SortParam :return: list of releases :rtype: list of JSON release objects :raises SW360Error: if there is a negative HTTP response @@ -64,32 +113,40 @@ def get_releases_by_name(self, name: str) -> List[Any]: if not name: raise SW360Error(message="No release name provided!") - full_url = self.url + "resource/api/releases?name=" + name - resp = self.api_get(full_url) - if resp and ("_embedded" in resp) and ("sw360:releases" in resp["_embedded"]): - return resp["_embedded"]["sw360:releases"] + fullbase_url = self.url + "resource/api/releases" + params = {"name": name} - return [] + url_with_param = self._add_params(fullbase_url, params) + + if sort is None: + sort = ReleaseSortColumn.SCORE.asc() + + return self.__get_releases_filtered(url_with_param, page, page_size, + sort) # return type List[Dict[str, Any]] | Optional[Dict[str, Any]] for Python 3.11 is good, # Union[List[Dict[str, Any]], Optional[Dict[str, Any]]] for lower Python versions is not good - def get_all_releases(self, fields: str = "", all_details: bool = False, - isNewClearingWithSourceAvailable: bool = False, page: int = -1, - page_size: int = -1, sort: str = "") -> Any: + def get_all_releases( + self, fields: str = "", all_details: bool = False, + isNewClearingWithSourceAvailable: bool = False, page: int = -1, + page_size: int = -1, sort: Optional[SortParam] = None + ) -> Any: """Get information of about all releases API endpoint: GET /releases - :param all_details: retrieve all project details (optional)) + :param fields: fields to include + :type fields: string + :param all_details: retrieve all release details (optional) :type all_details: bool - :param isNewClearingWithSourceAvailable: retrieve releases in new clearning state with source avail + :param isNewClearingWithSourceAvailable: retrieve releases in new clearing state with source available :type isNewClearingWithSourceAvailable: 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 releases ("name,desc"; "name,asc") - :type sort: str + :param sort: sort order for the releases (Sort by name if `None`) + :type sort: SortParam :return: list of releases :rtype: list of JSON release objects :raises SW360Error: if there is a negative HTTP response @@ -106,31 +163,23 @@ def get_all_releases(self, fields: str = "", all_details: bool = False, if fields: params["fields"] = fields - if page > -1: - params["page"] = str(page) - params["page_entries"] = str(page_size) - - if sort: - params["sort"] = sort - - full_url = self._add_params(fullbase_url, params) - - resp = self.api_get(full_url) + url_with_param = self._add_params(fullbase_url, params) - if page == -1 and resp and ("_embedded" in resp) and ("sw360:releases" in resp["_embedded"]): - return resp["_embedded"]["sw360:releases"] + if sort is None: + sort = ReleaseSortColumn.NAME.asc() - return resp + return self.__get_releases_filtered(url_with_param, page, page_size, + sort) def get_releases_by_external_id(self, ext_id_name: str, ext_id_value: str = "") -> List[Dict[str, Any]]: """Get releases by external id. `ext_id_value` can be left blank to search for all releases with `ext_id_name`. - API endpoint: GET /releases + API endpoint: GET /releases/searchByExternalIds :param ext_id_name: the name of the external id to look for - :param ext_id_value: the value of the external id to look for :type ext_id_name: string + :param ext_id_value: the value of the external id to look for :type ext_id_value: string :return: list of releases :rtype: list of JSON release objects @@ -139,11 +188,14 @@ def get_releases_by_external_id(self, ext_id_name: str, ext_id_value: str = "") if not ext_id_name: raise SW360Error(message="No external id name provided!") - resp = self.api_get( - self.url - + "resource/api/releases/searchByExternalIds?" - + ext_id_name + "=" + ext_id_value + url = self._add_params( + self.url + "resource/api/releases/searchByExternalIds", + { + ext_id_name: ext_id_value + } ) + + resp = self.api_get(url) if resp and ("_embedded" in resp) and ("sw360:releases" in resp["_embedded"]): return resp["_embedded"]["sw360:releases"] @@ -291,7 +343,7 @@ def get_users_of_release(self, release_id: str) -> Optional[Dict[str, Any]]: def link_packages_to_release(self, release_id: str, packages: List[str]) -> Optional[Dict[str, Any]]: """Link (new) packages to a given release. - API endpoint PATCH /release/{pid}/packages{rid} + API endpoint PATCH /release/{rid}/link/packages :param release_id: the id of the existing release :type release_id: string @@ -345,3 +397,33 @@ def get_recent_releases(self) -> Optional[List[Dict[str, Any]]]: return resp["_embedded"]["sw360:releases"] return [] + + def upload_attachment_to_release( + self, release_id: str, upload_file: str, upload_type: str = "SOURCE", + upload_comment: str = "" + ) -> Optional[Dict[str, Any]]: + """Upload an attachment to a given Release. + + API endpoint: POST /attachments & PATCH /releases/{id} + + :param release_id: the id of the Release + :type release_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 release id is missing or there is a negative HTTP response + """ + if not release_id: + raise SW360Error(message="No release id provided!") + + attachment_content = self._upload_resource_file(upload_file, upload_type, upload_comment) + attachment_content['attachmentType'] = upload_type # Make sure the type is correct + attachment_content['createdComment'] = upload_comment # Override + + current_release = self.get_release(release_id) + attachments = self._get_attachments(current_release) + attachments.append(attachment_content) + return self.update_release({'attachments': attachments}, release_id) diff --git a/tests/test_sw360_releases.py b/tests/test_sw360_releases.py index d4191e5..d6974de 100644 --- a/tests/test_sw360_releases.py +++ b/tests/test_sw360_releases.py @@ -17,6 +17,7 @@ sys.path.insert(1, "..") from sw360 import SW360, SW360Error # noqa: E402 +from sw360.sorting import ReleaseSortColumn # noqa: E402 class Sw360TestReleases(unittest.TestCase): @@ -161,7 +162,7 @@ def test_get_all_releases_isnewclearing_with_source_available(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?isNewClearingWithSourceAvailable=true", + url=self.MYURL + "resource/api/releases?isNewClearingWithSourceAvailable=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0"}]}}', status=200, content_type="application/json", @@ -185,7 +186,7 @@ def test_get_all_releases_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?allDetails=true", + url=self.MYURL + "resource/api/releases?allDetails=true&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", @@ -207,7 +208,7 @@ def test_get_all_releases_with_fields(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?fields=releaseDate", + url=self.MYURL + "resource/api/releases?fields=releaseDate&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", @@ -229,7 +230,7 @@ def test_get_all_releases_with_fields_and_all_details(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?allDetails=true&fields=releaseDate", + url=self.MYURL + "resource/api/releases?allDetails=true&fields=releaseDate&luceneSearch=true&page=0&page_entries=50&sort=name,asc", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", @@ -251,19 +252,18 @@ def test_get_all_releases_with_paging_and_sorting(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?page=2&page_entries=5&sort=name%2Casc", + url=self.MYURL + "resource/api/releases?page=2&page_entries=5&sort=name%2Casc&luceneSearch=true", body='{"_embedded": {"sw360:releases": [{"name": "Tethys.Logging", "version": "1.3.0", "releaseDate": "2018-03-04"}]}}', # noqa status=200, content_type="application/json", adding_headers={"Authorization": "Token " + self.MYTOKEN}, ) - releases = lib.get_all_releases(page=2, page_size=5, sort="name,asc") + releases = lib.get_all_releases(page=2, page_size=5, sort=ReleaseSortColumn.NAME.asc()) self.assertIsNotNone(releases) self.assertTrue(len(releases) > 0) - rel = releases["_embedded"]["sw360:releases"] - self.assertEqual("Tethys.Logging", rel[0]["name"]) - self.assertEqual("1.3.0", rel[0]["version"]) + self.assertEqual("Tethys.Logging", releases[0]["name"]) + self.assertEqual("1.3.0", releases[0]["version"]) @responses.activate def test_get_releases_by_external_id(self) -> None: @@ -316,7 +316,7 @@ def test_get_releases_by_name(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?name=john", + url=self.MYURL + "resource/api/releases?name=john&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_embedded": {"sw360:releases": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/releases/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", @@ -338,7 +338,7 @@ def test_get_releases_by_name_invalid_answer(self) -> None: responses.add( method=responses.GET, - url=self.MYURL + "resource/api/releases?name=john", + url=self.MYURL + "resource/api/releases?name=john&luceneSearch=true&page=0&page_entries=50&sort=score,asc", body='{"_xxembedded": {"sw360:releases": [{"name": "john", "version": "2.2.2", "_links": {"self": {"href": "https://my.server.com/resource/api/releases/08ddfd57636c4c47f4c879515007081f"}}}]}}', # noqa status=200, content_type="application/json", From d8067009a7549ba3dafeb455c5ba303c601f10de Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Sun, 16 Aug 2026 21:18:28 +0530 Subject: [PATCH 8/9] feat(vendors): update API calls to v20 style Signed-off-by: Gaurav Mishra --- sw360/sorting.py | 5 +++ sw360/vendor.py | 100 ++++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 95 insertions(+), 10 deletions(-) diff --git a/sw360/sorting.py b/sw360/sorting.py index 5a73092..99cf617 100644 --- a/sw360/sorting.py +++ b/sw360/sorting.py @@ -74,3 +74,8 @@ class PackageSortColumn(BaseSortMixin, Enum): NAME = "name" VERSION = "version" PACKAGE_MANAGER = "packageManager" + +class VendorSortColumn(BaseSortMixin, Enum): + SCORE = "score" + FULL_NAME = "fullName" + SHORT_NAME = "shortName" diff --git a/sw360/vendor.py b/sw360/vendor.py index 0d25109..8906cc8 100644 --- a/sw360/vendor.py +++ b/sw360/vendor.py @@ -12,21 +12,86 @@ from typing import Any, Dict, List, Optional from .base import BaseMixin +from .sorting import SortParam, VendorSortColumn from .sw360error import SW360Error class VendorMixin(BaseMixin): - def get_all_vendors(self) -> List[Dict[str, Any]]: + def get_all_vendors( + self, page: int = -1, page_size: int = -1, sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: """Returns all vendors API endpoint: GET /vendors - :return: a vendor - :rtype: JSON vendor object + :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 vendors (Sort by name if `None`) + :type sort: SortParam + :return: list of vendors + :rtype: list of JSON vendor objects :raises SW360Error: if there is a negative HTTP response """ - resp = self.api_get(self.url + "resource/api/vendors") + fullbase_url = self.url + "resource/api/vendors" + + if sort is None: + sort = VendorSortColumn.SHORT_NAME.asc() + + full_url = fullbase_url + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_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 resp and "_embedded" in resp and "sw360:vendors" in resp["_embedded"]: + return resp["_embedded"]["sw360:vendors"] + + return [] + + def search_vendors( + self, search_text: str, page: int = -1, page_size: int = -1, + sort: Optional[SortParam] = None + ) -> List[Dict[str, Any]]: + """Search vendors by full name or short name + + API endpoint: GET /vendors?searchText={search_text} + + :param search_text: search text + :type search_text: 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 vendors (Sort by score if `None`) + :type sort: SortParam + :return: list of vendors + :rtype: list of JSON vendor objects + :raises SW360Error: if there is a negative HTTP response + """ + if not search_text: + raise SW360Error(message="No search text provided!") + + fullbase_url = self.url + "resource/api/vendors" + params = {"searchText": search_text} + + if sort is None: + sort = VendorSortColumn.SCORE.asc() + + full_url = self._add_params(fullbase_url, params) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_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 resp and "_embedded" in resp and "sw360:vendors" in resp["_embedded"]: return resp["_embedded"]["sw360:vendors"] @@ -72,15 +137,15 @@ def create_new_vendor(self, vendor: Dict[str, Any]) -> 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 + API endpoint: PATCH /vendors/{id} + :param vendor: the new vendor data + :type vendor: JSON vendor object :param vendor_id: the id of the vendor to be updated :type vendor_id: string :raises SW360Error: if there is a negative HTTP response """ - # 2019-04-03: error 405 - not allowed - if not vendor_id: raise SW360Error(message="No vendor id provided!") @@ -90,15 +155,13 @@ def update_vendor(self, vendor: Dict[str, Any], vendor_id: str) -> Optional[Dict def delete_vendor(self, vendor_id: str) -> Optional[Dict[str, Any]]: """Delete an existing vendor - API endpoint: DELETE /vendors + API endpoint: DELETE /vendors/{id} :param vendor_id: the id of the vendor :type vendor_id: string :raises SW360Error: if there is a negative HTTP response """ - # 2019-04-03/2025-01-23: error 405 - not allowed - if not vendor_id: raise SW360Error(message="No vendor id provided!") @@ -111,3 +174,20 @@ def delete_vendor(self, vendor_id: str) -> Optional[Dict[str, Any]]: return response.json() return None + + def get_users_of_vendor(self, vendor_id: str) -> Optional[Dict[str, Any]]: + """Get information of about the releases supplied by the vendor + + API endpoint: GET /vendors/{id}/releases + + :param vendor_id: the id of the vendor to be requested + :type vendor_id: string + :return: all releases supplied by this vendor + :rtype: JSON objects + :raises SW360Error: if there is a negative HTTP response + """ + if not vendor_id: + raise SW360Error(message="No vendor id provided!") + + resp = self.api_get(self.url + "resource/api/vendors/" + vendor_id + "/releases") + return resp From e24ff851cb68bce6697909cb9fa14b2c3dbb6e3a Mon Sep 17 00:00:00 2001 From: Gaurav Mishra Date: Sun, 16 Aug 2026 21:18:44 +0530 Subject: [PATCH 9/9] feat(vuln): update API calls to v20 style Signed-off-by: Gaurav Mishra --- sw360/sorting.py | 7 +++++++ sw360/vulnerabilities.py | 40 ++++++++++++++++++++++++++++++++++++++-- 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/sw360/sorting.py b/sw360/sorting.py index 99cf617..a8358a4 100644 --- a/sw360/sorting.py +++ b/sw360/sorting.py @@ -79,3 +79,10 @@ class VendorSortColumn(BaseSortMixin, Enum): SCORE = "score" FULL_NAME = "fullName" SHORT_NAME = "shortName" + +class VulnerabilitySortColumn(BaseSortMixin, Enum): + LAST_UPDATE_DATE = "lastUpdateDate" + EXTERNAL_ID = "externalId" + TITLE = "title" + CVSS = "cvss" + PUBLISH_DATE = "publishDate" diff --git a/sw360/vulnerabilities.py b/sw360/vulnerabilities.py index 33516f8..616ba98 100644 --- a/sw360/vulnerabilities.py +++ b/sw360/vulnerabilities.py @@ -12,20 +12,56 @@ from typing import Any, Dict, Optional from .base import BaseMixin +from .sorting import SortParam, VulnerabilitySortColumn class VulnerabilitiesMixin(BaseMixin): - def get_all_vulnerabilities(self) -> Optional[Dict[str, Any]]: + def get_all_vulnerabilities( + self, search_text: str = "", all_details: bool = False, + page: int = 0, page_size: int = 10, sort: Optional[SortParam] = None + ) -> Optional[Dict[str, Any]]: """Get information of about all vulnerabilities API endpoint: GET /vulnerabilities + :param search_text: filter by externalId or title of vulnerability, + empty to get all + :type search_text: string + :param all_details: return all details about vulnerabilities ( + including releases) + :type all_details: bool + :param page: page to retrieve + :type page: int + :param page_size: page size to use + :type page_size: int + :param sort: sort order for the vulnerabilities (Sort by last update if + `None`) + :type sort: SortParam :return: list of vulnerabilities :rtype: list of JSON vulnerability objects :raises SW360Error: if there is a negative HTTP response """ + fullbase_url = self.url + "resource/api/vulnerabilities" + params = {} + + if search_text != "": + params["search"] = search_text + + if all_details: + params["allDetails"] = "true" + + if sort is None: + sort = VulnerabilitySortColumn.LAST_UPDATE_DATE.desc() + + full_url = self._add_params(fullbase_url, params) + if page > -1 and page_size > -1: + full_url = self._add_pagination(full_url, page, page_size, sort) + + if page_size == -1: + resp = self.api_get_all(full_url, sort) + else: + resp = self.api_get(full_url) - resp = self.api_get(self.url + "resource/api/vulnerabilities") return resp def get_vulnerability(self, vulnerability_id: str) -> Optional[Dict[str, Any]]: