Merge pull request #33 from sw360/32-add-package-api-support

add package api support

Author: tngraf

.gitignore

@@ -20,7 +20,10 @@ Test.tar.gz
 Test_file.zip
 test_all_components.json
 test_all_releases.json
+test_all_packages.json
+test_all_packages_with_details.json
 real_instance_tests.py
+sbom_to_packages.py
 
 # test coverage
 htmlcov/

ChangeLog.md

@@ -5,6 +5,12 @@
 
 # SW360 Base Library for Python
 
+## NEXT (V1.6.0)
+
+* packages REST API calls implemented.
+* unit test `test_login_failed_invalid_url` disabled because it delays all tests.
+* have unit tests for packages.
+
 ## V1.5.1
 
 * update requests 2.31.0 => 2.32.2 to fix CVE-2024-35195.

pyproject.toml

@@ -72,4 +72,4 @@ warn_unused_ignores         = true
 no_implicit_reexport        = true
 
 [tool.codespell]
-skip = "test_all_components.json,test_all_releases.json,./htmlcov/*,./__internal__/*,./docs/_static/*,./docs/searchindex.js"
+skip = "test_all_components.json,test_all_releases.json,./htmlcov/*,./__internal__/*,./docs/_static/*,./docs/searchindex.js,./docs/objects.inv"

sw360/base.py

@@ -136,7 +136,7 @@ def api_post(
 
         raise SW360Error(response, url)
 
-    def api_patch(self, url: str = "", json: Dict[str, Any] = {}) -> Optional[Dict[str, Any]]:
+    def api_patch(self, url: str = "", json: Any = {}) -> Optional[Dict[str, Any]]:
         """
         Send a PATCH request to the specified URL with the provided json data.
 
@@ -160,7 +160,10 @@ def api_patch(self, url: str = "", json: Dict[str, Any] = {}) -> Optional[Dict[s
         if response.ok:
             if response.status_code == 204:  # 204 = no content
                 return None
-            return response.json()
+            if response.content:
+                return response.json()
+            else:
+                return None
 
         raise SW360Error(response, url)
 

sw360/packages.py

@@ -0,0 +1,204 @@
+# -------------------------------------------------------------------------------
+# Copyright (c) 2024 Siemens
+# All Rights Reserved.
+# Authors: thomas.graf@siemens.com
+#
+# Licensed under the MIT license.
+# SPDX-License-Identifier: MIT
+# -------------------------------------------------------------------------------
+
+from typing import Any, Dict, List, Optional
+
+from .base import BaseMixin
+from .sw360error import SW360Error
+
+
+class PackagesMixin(BaseMixin):
+    def get_package(self, package_id: str) -> Optional[Dict[str, Any]]:
+        """Get information of about a package
+
+        API endpoint: GET /package/{id}
+
+        :param package_id: the id of the package to be requested
+        :type package_id: string
+        :return: a package
+        :rtype: JSON package object
+        :raises SW360Error: if there is a negative HTTP response
+        """
+        resp = self.api_get(self.url + "resource/api/packages/" + package_id)
+        return resp
+
+    def get_packages_by_name(self, name: str) -> List[Any]:
+        """Gets a list of packages that match the given name.
+
+        API endpoint: GET /packages?name=
+
+        :param name: the name
+        :type name: string
+        :return: list of packages
+        :rtype: list of JSON package objects
+        :raises SW360Error: if there is a negative HTTP response
+        """
+        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"]
+
+        return []
+
+    # 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_packages(self, fields: str = "", all_details: bool = False, page: int = -1,
+                         page_size: int = -1, sort: str = "") -> Any:
+        """Get information of about all packages
+
+        API endpoint: GET /releases
+
+        :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
+        :type page_size: int
+        :param sort: sort order for the packages ("name,desc"; "name,asc")
+        :type sort: str
+        :return: list of packages
+        :rtype: list of JSON package objects
+        :raises SW360Error: if there is a negative HTTP response
+        """
+        full_url = self.url + "resource/api/packages"
+        if all_details:
+            full_url = self._add_param(full_url, "allDetails=true")
+
+        if fields:
+            full_url = self._add_param(full_url, "fields=" + fields)
+
+        if page > -1:
+            full_url = self._add_param(full_url, "page=" + str(page))
+            full_url = self._add_param(full_url, "page_entries=" + str(page_size))
+
+        if sort:
+            # ensure HTML encoding
+            sort = sort.replace(",", "%2C")
+            full_url = self._add_param(full_url, "sort=" + sort)
+
+        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"]
+
+        return resp
+
+    def get_packages_by_packagemanager(self, manager: str, page: int = -1,
+                                       page_size: int = -1, sort: str = "") -> Any:
+        """Get information of about all packages of a specific package manager
+
+        API endpoint: GET /releases
+
+        :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
+        :type page_size: int
+        :param sort: sort order for the packages ("name,desc"; "name,asc")
+        :type sort: str
+        :return: list of packages
+        :rtype: list of JSON package objects
+        :raises SW360Error: if there is a negative HTTP response
+        """
+        full_url = self.url + "resource/api/packages"
+        full_url = self._add_param(full_url, "packageManager=" + str(manager))
+
+        if page > -1:
+            full_url = self._add_param(full_url, "page=" + str(page))
+            full_url = self._add_param(full_url, "page_entries=" + str(page_size))
+
+        if sort:
+            # ensure HTML encoding
+            sort = sort.replace(",", "%2C")
+            full_url = self._add_param(full_url, "sort=" + sort)
+
+        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"]
+
+        return resp
+
+    def create_new_package(self, name: str, version: str, purl: str,
+                           package_type: str, package_details: Dict[str, Any] = {}) -> Optional[Dict[str, Any]]:
+        """Create a new package
+
+        API endpoint: POST /packages
+
+        :param name: name of new package (usually set to component name)
+        :param version: version string of new package (e.g. "1.0")
+        :param purl: purl / package-url of the package
+        :param package_type: CycloneDX package type of the package
+        :param package_details: further package details as defined by SW360 REST API
+        :type name: string
+        :type version: string
+        :type purl: string
+        :type package_type: string
+        :type package_details: dict
+        :return: SW360 result
+        :rtype: JSON SW360 result object
+        :raises SW360Error: if there is a negative HTTP response
+        """
+
+        for param in "name", "version":
+            package_details[param] = locals()[param]
+        package_details["purl"] = purl
+        package_details["packageType"] = package_type
+
+        url = self.url + "resource/api/packages"
+        response = self.api_post(
+            url, json=package_details)
+        if response is not None:
+            if response.ok:
+                return response.json()
+        return None
+
+    def update_package(self, package: Dict[str, Any], package_id: str) -> Optional[Dict[str, Any]]:
+        """Update an existing package
+
+        API endpoint: PATCH /packages
+
+        :param release: the new package data
+        :param release_id: the id of the package to be updated
+        :type package: JSON
+        :type package_id: string
+        :return: SW360 result
+        :rtype: JSON SW360 result object
+        :raises SW360Error: if there is a negative HTTP response
+        """
+
+        if not package_id:
+            raise SW360Error(message="No package id provided!")
+
+        url = self.url + "resource/api/packages/" + package_id
+        return self.api_patch(url, json=package)
+
+    def delete_package(self, package_id: str) -> Optional[Dict[str, Any]]:
+        """Delete an existing package
+
+        API endpoint: DELETE /packages
+
+        :param package_id: the id of the package to be deleted
+        :type package_id: string
+        :return: SW360 result
+        :rtype: JSON SW360 result object
+        :raises SW360Error: if there is a negative HTTP response
+        """
+
+        if not package_id:
+            raise SW360Error(message="No package id provided!")
+
+        url = self.url + "resource/api/packages/" + package_id
+        response = self.api_delete(url)
+        if response is not None:
+            if response.ok:
+                if response.text:
+                    return response.json()
+        return None

sw360/project.py

@@ -518,3 +518,39 @@ def update_project_release_relationship(
 
         url = self.url + "resource/api/projects/" + project_id + "/release/" + release_id
         return self.api_patch(url, json=relation)
+
+    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}
+
+        :param project_id: the id of the existing project
+        :type project_id: string
+        :param packages: list of package ids
+        :type packages: list of string
+        :rtype: JSON SW360 result object
+        :raises SW360Error: if the project id is missing ir there is a negative HTTP response
+        """
+        if not project_id:
+            raise SW360Error(message="No project id provided!")
+
+        url = self.url + "resource/api/projects/" + project_id + "/link/packages/"
+        return self.api_patch(url, json=packages)
+
+    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}
+
+        :param project_id: the id of the existing project
+        :type project_id: string
+        :param packages: list of package ids
+        :type packages: list of string
+        :rtype: JSON SW360 result object
+        :raises SW360Error: if the project id is missing ir there is a negative HTTP response
+        """
+        if not project_id:
+            raise SW360Error(message="No project id provided!")
+
+        url = self.url + "resource/api/projects/" + project_id + "/unlink/packages/"
+        return self.api_patch(url, json=packages)

sw360/releases.py

@@ -1,5 +1,5 @@
 # -------------------------------------------------------------------------------
-# Copyright (c) 2019-2023 Siemens
+# Copyright (c) 2019-2024 Siemens
 # Copyright (c) 2022 BMW CarIT GmbH
 # All Rights Reserved.
 # Authors: thomas.graf@siemens.com, gernot.hillier@siemens.com
@@ -250,3 +250,39 @@ def get_users_of_release(self, release_id: str) -> Optional[Dict[str, Any]]:
 
         resp = self.api_get(self.url + "resource/api/releases/usedBy/" + release_id)
         return resp
+
+    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}
+
+        :param release_id: the id of the existing release
+        :type release_id: string
+        :param packages: list of package ids
+        :type packages: list of string
+        :rtype: JSON SW360 result object
+        :raises SW360Error: if the release id is missing ir there is a negative HTTP response
+        """
+        if not release_id:
+            raise SW360Error(message="No release id provided!")
+
+        url = self.url + "resource/api/releases/" + release_id + "/link/packages/"
+        return self.api_patch(url, json=packages)
+
+    def unlink_packages_from_release(self, release_id: str, packages: List[str]) -> Optional[Dict[str, Any]]:
+        """Unlink packages from a given release.
+
+        API endpoint PATCH /release/{pid}/packages{rid}
+
+        :param release_id: the id of the existing release
+        :type release_id: string
+        :param packages: list of package ids
+        :type packages: list of string
+        :rtype: JSON SW360 result object
+        :raises SW360Error: if the project id is missing ir there is a negative HTTP response
+        """
+        if not release_id:
+            raise SW360Error(message="No release id provided!")
+
+        url = self.url + "resource/api/releases/" + release_id + "/unlink/packages/"
+        return self.api_patch(url, json=packages)

sw360/sw360_api.py

@@ -19,6 +19,7 @@
 from .clearing import ClearingMixin
 from .components import ComponentsMixin
 from .license import LicenseMixin
+from .packages import PackagesMixin
 from .project import ProjectMixin
 from .releases import ReleasesMixin
 from .sw360error import SW360Error
@@ -46,7 +47,8 @@ class SW360(
     ProjectMixin,
     ReleasesMixin,
     VendorMixin,
-    VulnerabilitiesMixin
+    VulnerabilitiesMixin,
+    PackagesMixin
 ):
     """Python interface to the Siemens SW360 platform
 

tests/test_sw360_base.py

@@ -56,7 +56,7 @@ def test_login(self) -> None:
         actual = lib.login_api()
         self.assertTrue(actual)
 
-    def test_login_failed_invalid_url(self) -> None:
+    def x_test_login_failed_invalid_url(self) -> None:
         lib = SW360(self.MYURL, self.MYTOKEN, False)
 
         have_backup = False