Feature/bulk operations support (#39)

* Adding bulk operations support.
* Removed obsolete import.
* Minor fixes.
* Updated changelog.

Author: chsou

CHANGELOG.md

@@ -1,5 +1,7 @@
 # Changelog
 
+* Adding support for bulk operations.
+
 ## Version 1.6.1
 
 *  Adding `c8y_tk` namespace to distribution.

c8y_api/_main_api.py

@@ -17,7 +17,7 @@
 from c8y_api.model.inventory import Inventory, DeviceInventory, DeviceGroupInventory
 from c8y_api.model.measurements import Measurements
 from c8y_api.model.notification2 import Subscriptions, Tokens
-from c8y_api.model.operations import Operations
+from c8y_api.model.operations import Operations, BulkOperations
 from c8y_api.model.tenant_options import TenantOptions
 
 
@@ -44,6 +44,7 @@ def __init__(self, base_url: str, tenant_id: str, username: str = None, password
         self.__events = Events(self)
         self.__alarms = Alarms(self)
         self.__operations = Operations(self)
+        self.__bulk_operations = BulkOperations(self)
         self.__tenant_options = TenantOptions(self)
         self.__notification2_subscriptions = Subscriptions(self)
         self.__notification2_tokens = Tokens(self)
@@ -118,6 +119,11 @@ def operations(self) -> Operations:
         """Provide access to the Operation API."""
         return self.__operations
 
+    @property
+    def bulk_operations(self) -> BulkOperations:
+        """Provide access to the BulkOperation API."""
+        return self.__bulk_operations
+
     @property
     def tenant_options(self) -> TenantOptions:
         """Provide access to the Tenant Options API."""

c8y_api/model/__init__.py

@@ -20,6 +20,7 @@
            'User', 'GlobalRole', 'InventoryRole', 'Users', 'GlobalRoles', 'InventoryRoles', 'InventoryRoleAssignment',
            'Permission', 'ReadPermission', 'WritePermission', 'AnyPermission',
            'Application',
+           'Operation', 'BulkOperation', 'Operations', 'BulkOperations',
            'ManagedObject', 'Device', 'DeviceGroup', 'Fragment', 'NamedObject',
            'Inventory', 'DeviceInventory', 'DeviceGroupInventory',
            'Identity', 'ExternalId', 'Binary', 'Binaries',

c8y_api/model/operations.py

@@ -11,19 +11,19 @@
 
 from c8y_api._base_api import CumulocityRestApi
 
-from c8y_api.model._base import CumulocityResource, ComplexObject, SimpleObject
+from c8y_api.model._base import CumulocityResource, ComplexObject, SimpleObject, _DictWrapper
 from c8y_api.model._parser import ComplexObjectParser
 from c8y_api.model._util import _DateUtil
 
 
 class Operation(ComplexObject):
-    """ Represents an instance of a operation object in Cumulocity.
+    """ Represents an instance of an operation object in Cumulocity.
 
     Instances of this class are returned by functions of the corresponding
     Operation API. Use this class to create new or update existing
     operation.
 
-    See also: https://cumulocity.com/api/10.11.0/#tag/Operations
+    See also: https://cumulocity.com/api/core/#tag/Operations
     """
 
     class Status:
@@ -131,7 +131,7 @@ class Operations(CumulocityResource):
     This class can be used for get, search for, create, update and
     delete operations within the Cumulocity database.
 
-    See also: https://cumulocity.com/api/10.11.0/#tag/Operations
+    See also: https://cumulocity.com/api/core/#tag/Operations
     """
 
     def __init__(self, c8y: CumulocityRestApi):
@@ -141,7 +141,7 @@ def get(self, operation_id: str | int) -> Operation:
         """ Read a specific operation from the database.
 
         params:
-            operation_id (str|int):  database ID of a operation
+            operation_id (str|int):  database ID of an operation
 
         Returns:
             Operation object
@@ -211,7 +211,7 @@ def get_all(self, agent_id: str = None, device_id: str = None, status: str = Non
         available results are read immediately and returned as list.
 
         Returns:
-            List of matching Measurement objects
+            List of matching Operation objects
         """
         return list(self.select(agent_id=agent_id, device_id=device_id, status=status, bulk_id=bulk_id,
                                 fragment=fragment, before=before, after=after, min_age=min_age, max_age=max_age,
@@ -274,3 +274,184 @@ def delete_by(self, agent_id: str = None, device_id: str = None, status: str = N
         # remove &page_number= from the end
         query = base_query[:base_query.rindex('&')]
         self.c8y.delete(query)
+
+
+class BulkOperation(ComplexObject):
+    """ Represents an instance of a bulk operation object in Cumulocity.
+
+    Instances of this class are returned by functions of the corresponding
+    Bulk Operation API. Use this class to create new or update existing
+    operation.
+
+    See also: https://cumulocity.com/api/core/#tag/Bulk-operations
+    """
+
+    class Status:
+        """Bulk Operation statuses."""
+        ACTIVE = 'ACTIVE'
+        IN_PROGRESS = 'IN_PROGRESS'
+        COMPLETED = 'COMPLETED'
+        DELETED = 'DELETED'
+
+    class GeneralStatus:
+        """Bulk Operation general statuses."""
+        SCHEDULED = 'PENDING'
+        EXECUTING = 'EXECUTING'
+        EXECUTING_WITH_ERRORS = 'EXECUTING_WITH_ERRORS'
+        SUCCESSFUL = 'SUCCESSFUL'
+        FAILED = 'FAILED'
+        CANCELED = 'CANCELED'
+        COMPLETED_SUCCESSFULLY = 'COMPLETED SUCCESSFULLY'
+        COMPLETED_WITH_FAILURES = 'COMPLETED WITH FAILURES'
+
+    # these need to be defined like this for the abstract super functions
+    _resource = '/devicecontrol/bulkoperations'
+    _parser = ComplexObjectParser({
+        '_u_group_id': 'groupId',
+        '_u_failed_parent_id': 'failedParentId',
+        '_u_start_time': 'startDate',
+        '_u_creation_ramp': 'creationRamp',
+        'status': 'status',
+        'general_status': 'generalStatus'}, [])
+    _accept = 'application/vnd.com.nsn.cumulocity.bulkoperation+json'
+
+    def __init__(self, c8y=None, group_id: str = None, failed_parent_id: str = None,
+                 start_time: str | datetime = None, creation_ramp: float = None,
+                 operation_prototype: dict = None, **kwargs):
+        """ Create a new Operation object.
+
+        Params:
+            c8y (CumulocityRestApi):  Cumulocity connection reference; needs
+                to be set for direct manipulation (create, delete)
+            device_id (str):  Device ID which this operation is for
+            kwargs:  All additional named arguments are interpreted as
+                custom fragments e.g. for data points.
+
+        Returns:
+            Operation object
+        """
+        super().__init__(c8y, operationPrototype=operation_prototype, **kwargs)
+        self._u_group_id = group_id
+        self._u_failed_parent_id = failed_parent_id
+        self._u_start_time = _DateUtil.ensure_timestring(start_time)
+        self._u_creation_ramp = creation_ramp
+        self.status: str | None = None
+        self.general_status: str | None = None
+
+    group_id = SimpleObject.UpdatableProperty('_u_group_id')
+    failed_parent_id = SimpleObject.UpdatableProperty('_u_failed_parent_id')
+    start_time = SimpleObject.UpdatableProperty('_u_start_time')
+    creation_ramp = SimpleObject.UpdatableProperty('_u_creation_ramp')
+
+    @property
+    def operation_prototype(self) -> _DictWrapper:
+        return self.operationPrototype
+
+    @operation_prototype.setter
+    def operation_prototype(self, fragment):
+        self.__setitem__('operationPrototype', fragment)
+
+    @property
+    def start_datetime(self) -> datetime:
+        """Convert the operation's start time to a Python datetime object.
+
+        Returns:
+            Standard Python datetime object for the operation's start time.
+        """
+        return super()._to_datetime(self._u_start_time)
+
+    @classmethod
+    def from_json(cls, json) -> BulkOperation:
+        """ Build a new Operation instance from Cumulocity JSON.
+
+        The JSON is assumed to be in the format as it is used by the
+        Cumulocity REST API.
+
+        Params:
+            json (dict):  JSON object (nested dictionary)
+                representing an operation within Cumulocity
+
+        Returns:
+            Operation object
+        """
+        obj = cls._from_json(json, BulkOperation())
+        return obj
+
+    def create(self) -> BulkOperation:
+        """ Store the Bulk Operation within the database.
+
+        Returns:  A fresh BulkOperation object representing what was
+            created within the database (including the ID).
+        """
+        return self._create()
+
+    def update(self) -> BulkOperation:
+        """Update the BulkOperation within the database.
+
+        Returns:  A fresh BulkOperation object representing the updated
+            object within the database (including the ID).
+        """
+        return super()._update()
+
+
+class BulkOperations(CumulocityResource):
+    """ A wrapper for the standard Bulk Operation API.
+
+    This class can be used for get, search for, create, update and
+    delete bulk operations within the Cumulocity database.
+
+    See also: https://cumulocity.com/api/core/#tag/Bulk-operations
+    """
+
+    def __init__(self, c8y: CumulocityRestApi):
+        super().__init__(c8y, 'devicecontrol/bulkoperations')
+
+    def get(self, operation_id: str | int) -> BulkOperation:
+        """ Read a specific bulk operation from the database.
+
+        params:
+            operation_id (str|int):  database ID of a bulk operation
+
+        Returns:
+            BulkOperation object
+
+        Raises:
+            KeyError: If the ID cannot be resolved.
+        """
+        operation = BulkOperation.from_json(self._get_object(operation_id))
+        operation.c8y = self.c8y  # inject c8y connection into instance
+        return operation
+
+    def select(self, limit: int = None, page_size: int = 1000) -> Generator[BulkOperation]:
+        """ Query the database for operations and iterate over the results.
+
+        This function is implemented in a lazy fashion - results will only be
+        fetched from the database as long there is a consumer for them.
+
+        All parameters are considered to be filters, limiting the result set
+        to objects which meet the filters' specification.  Filters can be
+        combined (within reason).
+
+        Params:
+            limit (int):  Limit the number of results to this number.
+            page_size (int):  Define the number of operations which are
+                read (and parsed in one chunk). This is a performance
+                related setting.
+
+        Returns:
+            Generator[BulkOperation]: Iterable of matching BulkOperation objects
+        """
+        base_query = self._build_base_query(page_size=page_size)
+        return super()._iterate(base_query, limit, BulkOperation.from_json)
+
+    def get_all(self, limit: int = None, page_size: int = 1000) -> List[BulkOperation]:
+        """ Query the database for bulk operations and return the results
+        as list.
+
+        This function is a greedy version of the select function. All
+        available results are read immediately and returned as list.
+
+        Returns:
+            List of matching BulkOperation objects
+        """
+        return list(self.select(limit=limit, page_size=page_size))

integration_tests/test_bulk_operations.py

@@ -0,0 +1,63 @@
+# Copyright (c) 2020 Software AG,
+# Darmstadt, Germany and/or Software AG USA Inc., Reston, VA, USA,
+# and/or its subsidiaries and/or its affiliates and/or their licensors.
+# Use, reproduction, transfer, publication or disclosure is prohibited except
+# as specifically provided for in your License Agreement with Software AG.
+
+import time
+
+from c8y_api import CumulocityApi
+from c8y_api.model import BulkOperation, DeviceGroup, Operation
+
+
+def test_CRU(live_c8y: CumulocityApi, sample_device):  # noqa
+    """Verify that basic creation, lookup and update of Operations works as expected."""
+
+    # (1) Create a device group for the sample device
+    group:DeviceGroup = DeviceGroup(live_c8y,
+                                    root=True,
+                                    name=sample_device.name + '_Group').create()
+    group.add_child_asset(sample_device)
+
+
+    # (2) create bulk operation
+    bulk:BulkOperation = BulkOperation(live_c8y,
+                          group_id=group.id,
+                          start_time='now',
+                          creation_ramp=1,
+                          operation_prototype={
+                              'description': f"Update firmware for device group '{group.name}'.",
+                              'c8y_FirmWare': {
+                                  'version': '1.0.0'
+                              }}
+                          ).create()
+
+    # wait for the bulk operation to be processed
+    time.sleep(5)
+
+    # (3) initially the status should be EXECUTING/COMPLETED as all
+    #     child operations should have been created but not completed
+    bulk = live_c8y.bulk_operations.get(bulk.id)
+    assert bulk.general_status == BulkOperation.GeneralStatus.EXECUTING
+    assert bulk.status == BulkOperation.Status.COMPLETED
+    assert bulk.progress.all == 1
+    assert bulk.progress.pending == 1
+
+    # (4) find child operations
+    op = live_c8y.operations.get_all(bulk_id=bulk.id)[0]
+    assert op.status == Operation.Status.PENDING
+
+    # (5) kill child operations
+    op.status = Operation.Status.FAILED
+    op.update()
+
+    # (5) bulk operation should now be in PENDING state
+    bulk = live_c8y.bulk_operations.get(bulk.id)
+    assert bulk.general_status in [BulkOperation.GeneralStatus.COMPLETED_WITH_FAILURES,
+                                   BulkOperation.GeneralStatus.FAILED]
+    assert bulk.progress.all == 1
+    assert bulk.progress.failed == 1
+
+    # (6) cleanup
+    #     The bulk operation cannot be deleted physically
+    group.delete()