Merge pull request #119 from fossology/feat/file-items

feat(items): prepare new item endpoints for the next release

.github/workflows/fossologytests.yml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   test-latest:
-    name: Integration Tests (latest Fossology - 4.3.0)
+    name: Integration Tests (latest Fossology - 4.4.0-rc1)
     runs-on: ubuntu-latest
 
     container:
@@ -20,7 +20,7 @@ jobs:
 
     services:
       fossology:
-        image: fossology/fossology:4.3.0
+        image: fossology/fossology:4.4.0-rc1
         ports:
           - 8081:80
         volumes:

fossology/__init__.py

@@ -10,6 +10,7 @@
 from fossology.exceptions import AuthenticationError, FossologyApiError
 from fossology.folders import Folders
 from fossology.groups import Groups
+from fossology.items import Items
 from fossology.jobs import Jobs
 from fossology.license import LicenseEndpoint
 from fossology.obj import Agents, ApiInfo, HealthInfo, User
@@ -79,7 +80,9 @@ def fossology_token(
         exit(f"Server {url} does not seem to be running or is unreachable: {error}")
 
 
-class Fossology(Folders, Groups, LicenseEndpoint, Uploads, Jobs, Report, Users, Search):
+class Fossology(
+    Folders, Groups, Items, LicenseEndpoint, Uploads, Jobs, Report, Users, Search
+):
 
     """Main Fossology API class
 
@@ -104,7 +107,7 @@ def __init__(self, url, token, name=None):
         self.users = list()
         self.folders = list()
 
-        self.api = f"{self.host}/api/v1"
+        self.api = f"{self.host}/api/v2"
         self.session = requests.Session()
         self.session.headers.update({"Authorization": f"Bearer {self.token}"})
         self.info = self.get_info()

fossology/enums.py

@@ -195,3 +195,14 @@ class PrevNextSelection(Enum):
 
     WITHLICENSES = "withLicenses"
     NOCLEARING = "noClearing"
+
+
+class CopyrightStatus(Enum):
+    """Status of the copyrights:
+
+    ACTIVE
+    INACTIVE
+    """
+
+    ACTIVE = "active"
+    INACTIVE = "inactive"

fossology/items.py

@@ -0,0 +1,253 @@
+# mypy: disable-error-code="attr-defined"
+# Copyright 2019 Siemens AG
+# SPDX-License-Identifier: MIT
+import json
+import logging
+
+from fossology.enums import CopyrightStatus, PrevNextSelection
+from fossology.exceptions import FossologyApiError
+from fossology.obj import (
+    FileInfo,
+    GetBulkHistory,
+    GetClearingHistory,
+    GetPrevNextItem,
+    Upload,
+)
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+
+class Items:
+    """Class dedicated to all "uploads...items" related endpoints"""
+
+    def item_info(
+        self,
+        upload: Upload,
+        item_id: int,
+    ) -> FileInfo:
+        """Get the info for a specific upload item
+
+        API Endpoint: GET /uploads/{id}/item/{itemId}/info
+
+        :param upload: the upload to get items from
+        :param item_id: the id of the item
+        :type upload: Upload
+        :type item_id: int,
+        :return: the file info for the specified item
+        :rtype: FileInfo
+        :raises FossologyApiError: if the REST call failed
+        """
+        response = self.session.get(
+            f"{self.api}/uploads/{upload.id}/item/{item_id}/info"
+        )
+
+        if response.status_code == 200:
+            return FileInfo.from_json(response.json())
+
+        elif response.status_code == 404:
+            description = f"Upload {upload.id} or item {item_id} not found"
+            raise FossologyApiError(description, response)
+        else:
+            description = f"API error while getting info for item {item_id} from upload {upload.uploadname}"
+            raise FossologyApiError(description, response)
+
+    def item_copyrights(
+        self,
+        upload: Upload,
+        item_id: int,
+        status: CopyrightStatus,
+    ) -> int:
+        """Get the total copyrights of the mentioned upload tree ID
+
+        API Endpoint: GET /uploads/{id}/item/{itemId}/totalcopyrights
+
+        :param upload: the upload to get items from
+        :param item_id: the id of the item
+        :param status: the status of the copyrights
+        :type upload: Upload
+        :type item_id: int,
+        :return: the total number of copyrights for the uploadtree item
+        :rtype: int
+        :raises FossologyApiError: if the REST call failed
+        """
+        response = self.session.get(
+            f"{self.api}/uploads/{upload.id}/item/{item_id}/totalcopyrights?status={status.value}"
+        )
+
+        if response.status_code == 200:
+            return response.json()["total_copyrights"]
+
+        elif response.status_code == 404:
+            description = f"Upload {upload.id} or item {item_id} not found"
+            raise FossologyApiError(description, response)
+        else:
+            description = f"API error while getting total copyrights for item {item_id} from upload {upload.uploadname}."
+            raise FossologyApiError(description, response)
+
+    def get_clearing_history(
+        self,
+        upload: Upload,
+        item_id: int,
+    ) -> list[GetClearingHistory]:
+        """Get the clearing history for a specific upload item
+
+        API Endpoint: GET /uploads/{id}/item/{itemId}/clearing-history
+
+        :param upload: the upload to get items from
+        :param item_id: the id of the item with clearing decision
+        :type upload: Upload
+        :type item_id: int,
+        :return: the clearing history for the specified item
+        :rtype: List[GetClearingHistory]
+        :raises FossologyApiError: if the REST call failed
+        :raises AuthorizationError: if the REST call is not authorized
+        """
+        response = self.session.get(
+            f"{self.api}/uploads/{upload.id}/item/{item_id}/clearing-history"
+        )
+
+        if response.status_code == 200:
+            clearing_history = []
+            for action in response.json():
+                clearing_history.append(GetClearingHistory.from_json(action))
+            return clearing_history
+
+        elif response.status_code == 404:
+            description = f"Upload {upload.id} or item {item_id} not found"
+            raise FossologyApiError(description, response)
+        else:
+            description = f"API error while getting clearing history for item {item_id} from upload {upload.uploadname}."
+            raise FossologyApiError(description, response)
+
+    def get_prev_next(
+        self, upload: Upload, item_id: int, selection: PrevNextSelection | None = None
+    ) -> GetPrevNextItem:
+        """Get the index of the previous and the next time for an upload
+
+        API Endpoint: GET /uploads/{id}/item/{itemId}/prev-next
+
+        :param upload: the upload to get items from
+        :param item_id: the id of the item with clearing decision
+        :param selection: tell Fossology server how to select prev-next item
+        :type upload: Upload
+        :type item_id: int
+        :type selection: str
+        :return: list of items for the clearing history
+        :rtype: List[GetPrevNextItem]
+        :raises FossologyApiError: if the REST call failed
+        :raises AuthorizationError: if the REST call is not authorized
+        """
+        params = {}
+        if selection:
+            params["selection"] = selection
+
+        response = self.session.get(
+            f"{self.api}/uploads/{upload.id}/item/{item_id}/prev-next", params=params
+        )
+
+        if response.status_code == 200:
+            return GetPrevNextItem.from_json(response.json())
+
+        elif response.status_code == 404:
+            description = f"Upload {upload.id} or item {item_id} not found"
+            raise FossologyApiError(description, response)
+        else:
+            description = f"API error while getting prev-next items for {item_id} from upload {upload.uploadname}."
+            raise FossologyApiError(description, response)
+
+    def get_bulk_history(
+        self,
+        upload: Upload,
+        item_id: int,
+    ) -> list[GetBulkHistory]:
+        """Get the bulk history for a specific upload item
+
+        API Endpoint: GET /uploads/{id}/item/{itemId}/bulk-history
+
+        :param upload: the upload to get items from
+        :param item_id: the id of the item with clearing decision
+        :type upload: Upload
+        :type item_id: int
+        :return: list of data from the bulk history
+        :rtype: List[GetBulkHistory]
+        :raises FossologyApiError: if the REST call failed
+        :raises AuthorizationError: if the REST call is not authorized
+        """
+        response = self.session.get(
+            f"{self.api}/uploads/{upload.id}/item/{item_id}/bulk-history"
+        )
+
+        if response.status_code == 200:
+            bulk_history = []
+            for item in response.json():
+                bulk_history.append(GetBulkHistory.from_json(item))
+            return bulk_history
+
+        elif response.status_code == 404:
+            description = f"Upload {upload.id} or item {item_id} not found"
+            raise FossologyApiError(description, response)
+        else:
+            description = f"API error while getting bulk history for {item_id} from upload {upload.uploadname}."
+            raise FossologyApiError(description, response)
+
+    def schedule_bulk_scan(
+        self,
+        upload: Upload,
+        item_id: int,
+        spec: dict,
+    ):
+        """Schedule a bulk scan for a specific upload item
+
+        API Endpoint: POST /uploads/{id}/item/{itemId}/bulk-scan
+
+        Bulk scan specifications `spec` are added to the request body,
+        following options are available:
+
+        >>> bulk_scan_spec = {
+        ...     "bulkActions": [
+        ...         {
+        ...             "licenseShortName": 'MIT',
+        ...             "licenseText": 'License text',
+        ...             "acknowledgement": 'Acknowledgment text',
+        ...             "comment": 'Comment text',
+        ...             "licenseAction": 'ADD', # or 'REMOVE'
+        ...         }
+        ...     ],
+        ...     "refText": 'Reference Text',
+        ...     "bulkScope": 'folder', # or upload
+        ...     "forceDecision": 'false',
+        ...     "ignoreIrre": 'false',
+        ...     "delimiters": 'DEFAULT',
+        ...     "scanOnlyFindings": 'true',
+        ... }
+
+        :param upload: the upload for the bulk scan
+        :param item_id: the id of the item for the bulk scan
+        :param spec: bulk scan specification
+        :type upload: Upload
+        :type item_id: int
+        :raises FossologyApiError: if the REST call failed
+        :raises AuthorizationError: if the REST call is not authorized
+        """
+        headers = {"Content-Type": "application/json"}
+        response = self.session.post(
+            f"{self.api}/uploads/{upload.id}/item/{item_id}/bulk-scan",
+            headers=headers,
+            data=json.dumps(spec),
+        )
+        if response.status_code == 201:
+            logger.info(
+                f"Bulk scan scheduled for upload {upload.uploadname}, item {item_id}"
+            )
+        elif response.status_code == 400:
+            description = (
+                f"Bad bulk scan request for upload {upload.id}, item {item_id}"
+            )
+            raise FossologyApiError(description, response)
+        elif response.status_code == 404:
+            description = f"Upload {upload.id} or item {item_id} not found"
+            raise FossologyApiError(description, response)
+        else:
+            description = f"API error while scheduling bulk scan for item {item_id} from upload {upload.uploadname}."
+            raise FossologyApiError(description, response)

fossology/obj.py

@@ -530,6 +530,50 @@ def from_json(cls, json_dict):
         return cls(**json_dict)
 
 
+class FileInfo(object):
+
+    """FOSSology file info response.
+
+    Represents a FOSSology file info response.
+
+    :param view_info: view info of the file
+    :param meta_info: meta info of the file
+    :param package_info: package info of the file
+    :param tag_info: tag info of the file
+    :param reuse_info: reuse info of the file
+    :param kwargs: handle any other license information provided by the fossology instance
+    :type view_info: Object
+    :type meta_info: Object
+    :type package_info: Object
+    :type tag_info: Object
+    :type reuse_info: Object
+    :type kwargs: key word argument
+    """
+
+    def __init__(
+        self,
+        view_info,
+        meta_info,
+        package_info,
+        tag_info,
+        reuse_info,
+        **kwargs,
+    ):
+        self.view_info = view_info
+        self.meta_info = meta_info
+        self.package_info = package_info
+        self.tag_info = tag_info
+        self.reuse_info = reuse_info
+        self.additional_info = kwargs
+
+    def __str__(self):
+        return f"File view {self.view_info}"
+
+    @classmethod
+    def from_json(cls, json_dict):
+        return cls(**json_dict)
+
+
 class Upload(object):
 
     """FOSSology upload.