Merge pull request #79 from Cumulocity-IoT/feature/inventory_get_by

Author: chsou

CHANGELOG.md

@@ -1,5 +1,6 @@
 # Changelog
 
+* Added `get_by` function to inventory API to return a single object by query.
 * Added `CumulocityApp` class to module `c8y_tk.app` which allows working with Cumulocity interactively, e.g. in a
   Jupyther notebook. It will deal with environment variables just like the other connection helpers but will also
   ask interactively for missing info, e.g. a second factor with 2FA. It also integrates well with the

c8y_api/model/_base.py

@@ -685,6 +685,11 @@ def _map_params(
         def multi(*xs):
             return sum(bool(x) for x in xs) > 1
 
+        def stringify(value):
+            if isinstance(value, bool):
+                return str(value).lower()
+            return value
+
         if multi(min_age, before, date_to):
             raise ValueError("Only one of 'min_age', 'before' and 'date_to' query parameters must be used.")
         if multi(max_age, after, date_from):
@@ -739,11 +744,11 @@ def multi(*xs):
             'createdTo': created_to,
             'lastUpdatedFrom': updated_from,
             'lastUpdatedTo': updated_to,
-            'withSourceAssets': with_source_assets,
-            'withSourceDevices': with_source_devices,
-            'revert': str(reverse).lower() if reverse is not None else None,
+            'withSourceAssets': stringify(with_source_assets),
+            'withSourceDevices': stringify(with_source_devices),
+            'revert': stringify(reverse),
             'pageSize': page_size}.items() if v is not None}
-        params.update({_StringUtil.to_pascal_case(k): v for k, v in kwargs.items() if v is not None})
+        params.update({_StringUtil.to_pascal_case(k): stringify(v) for k, v in kwargs.items() if v is not None})
         tuples = list(params.items())
         if series:
             if isinstance(series, list):
@@ -758,8 +763,9 @@ def _prepare_query(self, resource: str = None, expression: str = None, **kwargs)
             return resource or self.resource
         return (resource or self.resource) + '?' + encoded
 
-    def _get_object(self, object_id):
-        return self.c8y.get(self.build_object_path(object_id))
+    def _get_object(self, object_id, **kwargs):
+        query = self._prepare_query(self.build_object_path(object_id), **kwargs)
+        return self.c8y.get(query)
 
     def _get_page(self, base_query: str, page_number: int):
         sep = '&' if '?' in base_query else '?'

c8y_api/model/inventory.py

@@ -23,19 +23,46 @@ class Inventory(CumulocityResource):
     def __init__(self, c8y):
         super().__init__(c8y, 'inventory/managedObjects')
 
-    def get(self, id) -> ManagedObject:  # noqa (id)
+    def get(
+            self,
+            id: str,  # noqa
+            with_children: bool = None,
+            with_children_count: bool = None,
+            skip_children_names: bool = None,
+            with_parents: bool = None,
+            with_latest_values: bool = None,
+            **kwargs) -> ManagedObject:
         """ Retrieve a specific managed object from the database.
 
         Args:
-            ID of the managed object
+            id (str): Cumulocity ID of the managed object
+            with_children (bool):  Whether children with ID and name should be
+                included with each returned object
+            with_children_count (bool): When set to true, the returned result
+                will contain the total number of children in the respective
+                child additions, assets and devices sub fragments.
+            skip_children_names (bool):  If true, returned references of child
+                devices won't contain their names.
+            with_parents (bool): Whether to include a device's parents.
+            with_latest_values (bool):  If true the platform includes the
+                fragment `c8y_LatestMeasurements, which contains the latest
+                measurement values reported by the device to the platform.
 
         Returns:
              A ManagedObject instance
 
         Raises:
             KeyError:  if the ID is not defined within the database
         """
-        managed_object = ManagedObject.from_json(self._get_object(id))
+        managed_object = ManagedObject.from_json(self._get_object(
+            id,
+            with_children=with_children,
+            with_children_count=with_children_count,
+            skip_children_names=skip_children_names,
+            with_parents=with_parents,
+            with_latest_values=with_latest_values,
+            **kwargs)
+        )
         managed_object.c8y = self.c8y  # inject c8y connection into instance
         return managed_object
 
@@ -98,6 +125,66 @@ def get_all(
             as_values=as_values,
             **kwargs))
 
+    def get_by(
+            self,
+            expression: str = None,
+            query: str = None,
+            ids: list[str | int] = None,
+            order_by: str = None,
+            type: str = None,
+            parent: str = None,
+            fragment: str = None,
+            fragments: list[str] = None,
+            name: str = None,
+            owner: str = None,
+            text: str = None,
+            only_roots: str = None,
+            with_children: bool = None,
+            with_children_count: bool = None,
+            skip_children_names: bool = None,
+            with_groups: bool = None,
+            with_parents: bool = None,
+            with_latest_values: bool = None,
+            as_values: str | tuple | list[str | tuple] = None,
+            **kwargs) -> ManagedObject:
+        """ Query the database for a specific managed object.
+
+        This function is a special version of the `select` function assuming a single
+        result being returned by the query.
+
+        Returns:
+            A ManagedObject instance
+
+        Raises:
+            ValueError:  if the query did not return any or more than one result.
+        """
+        result = list(self.select(
+            expression=expression,
+            query=query,
+            ids=ids,
+            order_by=order_by,
+            type=type,
+            parent=parent,
+            fragment=fragment,
+            fragments=fragments,
+            name=name,
+            owner=owner,
+            text=text,
+            only_roots=only_roots,
+            with_children=with_children,
+            with_children_count=with_children_count,
+            skip_children_names=skip_children_names,
+            with_groups=with_groups,
+            with_parents=with_parents,
+            with_latest_values=with_latest_values,
+            page_size=2,
+            as_values=as_values,
+            **kwargs))
+        if len(result) == 1:
+            return result[0]
+        raise ValueError("No matching object found." if not result
+                         else "Ambiguous query; multiple matching objects found.")
+
     def get_count(
             self,
             expression: str = None,
@@ -455,19 +542,46 @@ def accept(self, id: str):  # noqa (id)
         """
         self.c8y.put('/devicecontrol/newDeviceRequests/' + str(id), {'status': 'ACCEPTED'})
 
-    def get(self, id: str) -> Device:  # noqa (id)
+    def get(
+            self,
+            id: str,  # noqa
+            with_children: bool = None,
+            with_children_count: bool = None,
+            skip_children_names: bool = None,
+            with_parents: bool = None,
+            with_latest_values: bool = None,
+            **kwargs) -> Device:
         """ Retrieve a specific device object.
 
         Args:
-            id (str): ID of the device object
+            id (str): Cumulocity ID of the device object
+            with_children (bool):  Whether children with ID and name should be
+                included with each returned object
+            with_children_count (bool): When set to true, the returned result
+                will contain the total number of children in the respective
+                child additions, assets and devices sub fragments.
+            skip_children_names (bool):  If true, returned references of child
+                devices won't contain their names.
+            with_parents (bool): Whether to include a device's parents.
+            with_latest_values (bool):  If true the platform includes the
+                fragment `c8y_LatestMeasurements, which contains the latest
+                measurement values reported by the device to the platform.
 
         Returns:
             A Device instance
 
         Raises:
             KeyError:  if the ID is not defined within the database
         """
-        device = Device.from_json(self._get_object(id))
+        device = Device.from_json(self._get_object(
+            id,
+            with_children=with_children,
+            with_children_count=with_children_count,
+            skip_children_names=skip_children_names,
+            with_parents=with_parents,
+            with_latest_values=with_latest_values,
+            **kwargs)
+        )
         device.c8y = self.c8y
         return device
 

c8y_api/model/managedobjects.py

@@ -286,7 +286,6 @@ def __init__(self, c8y: CumulocityRestApi = None,
         self.update_time = None
         self.child_devices = []
         self.child_assets = []
-        """List of NamedObject references to child assets."""
         self.child_additions = []
         self.parent_devices = []
         self.parent_assets = []
@@ -331,12 +330,16 @@ def _from_json(cls, json: dict, obj: Any) -> Any:
             mo.is_device = True
         if 'c8y_IsBinary' in json:
             mo.is_binary = True
-        if 'childDevices' in json:
-            mo.child_devices = cls._parse_references(json['childDevices'])
-        if 'childAssets' in json:
-            mo.child_assets = cls._parse_references(json['childAssets'])
-        if 'childAdditions' in json:
-            mo.child_additions = cls._parse_references(json['childAdditions'])
+        for fragment, field in [
+            ('childDevices', 'child_devices'),
+            ('childAssets', 'child_assets'),
+            ('childAdditions', 'child_additions'),
+            ('deviceParents', 'parent_devices'),
+            ('assetParents', 'parent_assets'),
+            ('additionParents', 'parent_additions'),
+        ]:
+            if fragment in json:
+                mo.__dict__[field] = cls._parse_references(json[fragment])
         return mo
 
     @classmethod

c8y_tk/app/subscription_listener.py

@@ -119,6 +119,8 @@ def listen(self):
 
         This is blocking.
         """
+        # pylint: disable=too-many-branches
+
         # safely invoke a callback function blocking or non-blocking
         def invoke_callback(callback, is_blocking, _, arg):
             def safe_invoke(a):
@@ -128,7 +130,7 @@ def safe_invoke(a):
                         self._log.debug(f"Invoking callback: {callback.__module__}.{callback.__name__}")
                     callback(a)
                 except Exception as callback_error:
-                    self._log.error(f"Uncaught exception in callback: {callback_error}", exc_info=error)
+                    self._log.error(f"Uncaught exception in callback: {callback_error}", exc_info=callback_error)
             if is_blocking:
                 safe_invoke(arg)
             else:
@@ -180,8 +182,8 @@ def safe_invoke(a):
                 # schedule next run, skip if already exceeded
                 next_run = time.monotonic() + self.polling_interval
                 if self._log.isEnabledFor(logging.DEBUG):
-                    next_run_datetime = (datetime.now(timezone.utc) + timedelta(seconds=self.polling_interval) ).isoformat(sep=' ', timespec='seconds')
-                    self._log.debug(f"Next run at {next_run_datetime}.")
+                    next_run_datetime = datetime.now(timezone.utc) + timedelta(seconds=self.polling_interval)
+                    self._log.debug(f"Next run at {next_run_datetime.isoformat(sep=' ', timespec='seconds')}.")
                 # sleep until next poll
                 if not time.monotonic() > next_run:
                     self._is_closed.wait(next_run - time.monotonic())
@@ -192,6 +194,7 @@ def safe_invoke(a):
             if self._executor:
                 self._executor.shutdown(wait=False, cancel_futures=False)
 
+        # pylint: disable=broad-exception-caught
         except Exception as error:
             self._log.error(f"Uncaught exception during listen: {error}", exc_info=error)
 

integration_tests/test_alarms.py

@@ -141,7 +141,6 @@ def sample_alarms(session_device, module_factory) -> List[Alarm]:
     ]
 
 
-
 def test_apply_by(live_c8y: CumulocityApi, session_device: Device):
     """Verify that the apply_by function works."""