Merge pull request #78 from Cumulocity-IoT/feature/better-get_last

Feature/better get last

Author: chsou

c8y_api/app/__init__.py

@@ -93,9 +93,9 @@ def get_item(key: str, dictionary: dict) -> str | None:
             info = get_item('Authorization', headers)
 
         if not info:
-
             keys = ", ".join([*headers.keys(), *cookies.keys()]) or "None"
             raise KeyError(f"Unable to resolve Authorization information. Found keys: {keys}.")
+
         return info
 
     @staticmethod

c8y_api/model/_util.py

@@ -50,6 +50,7 @@ def encode_odata_text_value(value):
         encoded_quotes = sub('\'', '\'\'', value)
         return encoded_quotes if " " not in encoded_quotes else f"'{encoded_quotes}'"
 
+
 class _DateUtil(object):
 
     @staticmethod

c8y_api/model/events.py

@@ -250,6 +250,16 @@ def _check_params(fragment, fragment_type, fragment_value):
         if fragment_value and not (fragment_type or fragment):
             raise ValueError("Fragment value filter also needs 'fragment_type' or 'fragment' filter.")
 
+    def _prepare_base_query(
+            self,
+            fragment: str = None,
+            fragment_type: str = None,
+            fragment_value: str = None,
+            **kwargs) -> str:
+        Events._check_params(fragment, fragment_type, fragment_value)
+        base_query = self._prepare_query(**kwargs)
+        return base_query
+
     def select(self,
                expression: str = None,
                type: str = None, source: str = None, fragment: str = None,  # noqa (type)
@@ -323,22 +333,26 @@ def select(self,
         Returns:
             Generator for Event objects
         """
-        Events._check_params(fragment, fragment_type, fragment_value)
-
-        base_query = self._prepare_query(
+        base_query = self._prepare_base_query(
             expression=expression,
-            type=type, source=source, fragment=fragment,
-            fragment_type=fragment_type, fragment_value=fragment_value,
-            before=before, after=after,
-            date_from=date_from, date_to=date_to,
-            created_before=created_before, created_after=created_after,
-            created_from=created_from, created_to=created_to,
-            updated_before=updated_before, updated_after=updated_after,
-            last_updated_from=last_updated_from, last_updated_to=last_updated_to,
-            min_age=min_age, max_age=max_age,
+            type=type,
+            source=source,
+            fragment=fragment,
+            fragment_type=fragment_type,
+            fragment_value=fragment_value,
+            before=before,
+            after=after,
+            created_before=created_before,
+            created_after=created_after,
+            updated_before=updated_before,
+            updated_after=updated_after,
+            min_age=min_age,
+            max_age=max_age,
+            date_from=date_from,
+            date_to=date_to,
             reverse=reverse,
-            with_source_devices=with_source_devices, with_source_assets=with_source_assets,
-            page_size=page_size,
+            with_source_assets=with_source_assets,
+            with_source_devices=with_source_devices,
             **kwargs)
         return super()._iterate(
             base_query,
@@ -415,29 +429,73 @@ def get_count(
         Returns:
             Number of potential results
         """
-        Events._check_params(
+        base_query = self._prepare_base_query(
+            expression=expression,
+            type=type,
+            source=source,
             fragment=fragment,
             fragment_type=fragment_type,
             fragment_value=fragment_value,
-        )
-        base_query = self._prepare_query(
-            expression=expression,
-            type=type, source=source, fragment=fragment,
-            fragment_type=fragment_type, fragment_value=fragment_value,
-            before=before, after=after,
-            date_from=date_from, date_to=date_to,
-            created_before=created_before, created_after=created_after,
-            created_from=created_from, created_to=created_to,
-            updated_before=updated_before, updated_after=updated_after,
-            last_updated_from=last_updated_from, last_updated_to=last_updated_to,
-            min_age=min_age, max_age=max_age,
+            before=before,
+            after=after,
+            created_before=created_before,
+            created_after=created_after,
+            updated_before=updated_before,
+            updated_after=updated_after,
+            min_age=min_age,
+            max_age=max_age,
+            date_from=date_from,
+            date_to=date_to,
             reverse=reverse,
-            with_source_devices=with_source_devices, with_source_assets=with_source_assets,
-            limit=limit,
-            **kwargs
-        )
+            with_source_assets=with_source_assets,
+            with_source_devices=with_source_devices,
+            **kwargs)
         return self._get_count(base_query)
 
+    def get_last(
+            self,
+            expression: str = None,
+            type: str = None, source: str = None, fragment: str = None,  # noqa (type)
+            fragment_type: str = None, fragment_value: str = None,
+            before: str | datetime = None, date_to: str | datetime = None,
+            created_before: str | datetime = None, created_to: str | datetime = None,
+            updated_before: str | datetime = None, last_updated_to: str | datetime = None,
+            min_age: timedelta = None,
+            with_source_assets: bool = None,
+            with_source_devices: bool = None,
+            **kwargs
+        ) -> Event | None:
+        """Retrieve the most recent event.
+        """
+        after = None
+        if not before and not date_to and not min_age:
+            after = '1970-01-01'
+        base_query = self._prepare_base_query(
+            expression=expression,
+            type=type,
+            source=source,
+            fragment=fragment,
+            fragment_type=fragment_type,
+            fragment_value=fragment_value,
+            before=before,
+            date_to=date_to,
+            after=after, # fallback if no other is defined
+            created_before=created_before,
+            created_to=created_to,
+            updated_before=updated_before,
+            last_updated_to=last_updated_to,
+            min_age=min_age,
+            reverse=False,  # newest first (non-standard)
+            with_source_assets=with_source_assets,
+            with_source_devices=with_source_devices,
+            **kwargs)
+        r = self._get_page(base_query, 1)
+        if not r:
+            return None
+        e = Event.from_json(r[0])
+        e.c8y = self.c8y  # inject c8y connection into instance
+        return e
+
     def create(self, *events: Event):
         """Create event objects within the database.
 

c8y_api/model/operations.py

@@ -274,14 +274,14 @@ def get_last(
             before: str | datetime = None,
             min_age: timedelta = None,
             **kwargs
-    ) -> Operation:
+    ) -> Operation|None:
         """ Query the database and return the last matching operation.
 
         This function is a special variant of the select function. Only
         the last matching result is returned.
 
         Returns:
-            Last matching Operation object
+            Last matching Operation object or None
         """
         # at least one date qualifier is required for this query to function,
         # so we enforce the 'after' filter if nothing else is specified
@@ -296,9 +296,12 @@ def get_last(
             reverse=True, page_size=1,
             **kwargs
         )
-        m = Operation.from_json(self._get_page(base_query, 1)[0])
-        m.c8y = self.c8y  # inject c8y connection into instance
-        return m
+        r = self._get_page(base_query, 1)
+        if not r:
+            return None
+        o = Operation.from_json(r[0])
+        o.c8y = self.c8y  # inject c8y connection into instance
+        return o
 
     def get_count(
             self,

c8y_tk/app/__init__.py

@@ -51,8 +51,6 @@ def __init__(
             startup_delay (float):  A minimum delay before a newly added
                 microservice is considered to be "added" (the callback
                 invocation will be delayed by this).
-
-
         """
         self._n = self._n + 1
         self._instance_name = f"{__name__}.{type(self).__name__}[{self._n}]"\
@@ -64,7 +62,7 @@ def __init__(
         self.callbacks = [(callback, blocking)] if callback else []
         self.callbacks_on_add = []
         self.callbacks_on_remove = []
-        self._log = logging.Logger(self._instance_name)
+        self._log = logging.getLogger(self._instance_name)
         self._executor = None
         self._callback_futures = set()
         self._listen_thread = None
@@ -121,10 +119,11 @@ def listen(self):
         This is blocking.
         """
         # safely invoke a callback function blocking or non-blocking
-        def invoke_callback(callback, is_blocking, arg):
+        def invoke_callback(callback, is_blocking, _, arg):
             def safe_invoke(a):
                 # pylint: disable=broad-exception-caught
                 try:
+                    self._log.debug(f"Invoking callback: {callback}")
                     callback(a)
                 except Exception as error:
                     self._log.error(f"Uncaught exception in callback: {error}", exc_info=error)
@@ -154,23 +153,24 @@ def safe_invoke(a):
             removed = last_subscribers - current_subscribers
             # run 'removed' callbacks
             for tenant_id in removed:
-                self._log.info(f"Tenant subscription removed: ${tenant_id}.")
+                self._log.info(f"Tenant subscription removed: {tenant_id}.")
                 for fun, blocking in self.callbacks_on_remove:
-                    invoke_callback(fun, blocking, tenant_id)
+                    invoke_callback(fun, blocking, 'Removed', tenant_id)
             # wait remaining time for startup delay
             if added and self.startup_delay:
                 min_startup_delay = self.startup_delay - (time.monotonic() - now)
                 if min_startup_delay > 0:
                     time.sleep(min_startup_delay)
             # run 'added' callbacks
             for tenant_id in added:
-                self._log.info(f"Tenant subscription added: ${tenant_id}.")
+                self._log.info(f"Tenant subscription added: {tenant_id}.")
                 for fun, blocking in self.callbacks_on_add:
-                    invoke_callback(fun, blocking, tenant_id)
+                    invoke_callback(fun, blocking, 'Added', tenant_id)
             # run 'any' callbacks
             if added or removed:
+                self._log.info(f"Tenant subscriptions changed: {current_subscribers}.")
                 for fun, blocking in self.callbacks:
-                    invoke_callback(fun, blocking, current_subscribers)
+                    invoke_callback(fun, blocking, None, current_subscribers)
             # set new baseline
             last_subscribers = current_subscribers
             # schedule next run, skip if already exceeded

integration_tests/test_events.py

@@ -1,9 +1,10 @@
 # Copyright (c) 2025 Cumulocity GmbH
 
-# pylint: disable=redefined-outer-name
+# pylint: disable=redefined-outer-name, protected-access
 
 from __future__ import annotations
 
+import datetime as dt
 import logging
 import os
 import tempfile
@@ -14,6 +15,7 @@
 
 from c8y_api import CumulocityApi
 from c8y_api.model import Event, Device
+from c8y_api.model._util import _DateUtil
 from util.testing_util import RandomNameGenerator
 
 
@@ -37,9 +39,10 @@ def sample_events(factory, sample_device) -> List[Event]:
     be removed after the test function."""
     typename = RandomNameGenerator.random_name()
     result = []
+    now = _DateUtil.now()
     for i in range(1, 6):
         event = Event(type=f'{typename}_{i}', text=f'{typename} text', source=sample_device.id,
-                      time='2020-12-31T11:33:55Z')
+                      time=now + dt.timedelta(minutes=i))
         result.append(factory(event))
     return result
 
@@ -151,19 +154,23 @@ def test_filter_by_update_time(live_c8y: CumulocityApi, sample_device, sample_ev
     pivot = updated_datetimes[len(updated_datetimes)//2]
 
     before_events = live_c8y.events.get_all(source=event.source, updated_before=pivot)
-    after_events = live_c8y.events.get_all(source=event.source, updated_after=pivot)
+    after_events = live_c8y.events.get_all(source=event.source, updated_after=pivot, reverse=True)
+    last_event_before = live_c8y.events.get_last(source=event.source, updated_before=pivot)
+    last_event_after = live_c8y.events.get_last(source=event.source, updated_after=pivot)
 
     # -> selected events should match the update times from 'before'
     # upper boundary, i.e. before/to timestamp is exclusive -> does not include pivot
     before_datetimes = list(filter(lambda x: x < pivot, updated_datetimes))
     result_datetimes = [a.updated_datetime for a in before_events]
     assert sorted(result_datetimes) == sorted(before_datetimes)
+    assert last_event_before.updated_datetime == max(before_datetimes)
 
     # -> selected events should match the update times from 'after'
     # lower boundary, i.e. after/from timestamp is inclusive -> includes pivot
     after_datetimes = list(filter(lambda x: x >= pivot, updated_datetimes))
     result_datetimes = [a.updated_datetime for a in after_events]
     assert sorted(result_datetimes) == sorted(after_datetimes)
+    assert last_event_after.updated_datetime == max(after_datetimes)
 
 
 def test_CRUD_attachments(live_c8y: CumulocityApi, sample_device: Device, sample_events: List[Event]):  # noqa (case)

integration_tests/test_operations.py

@@ -1,6 +1,5 @@
 # Copyright (c) 2025 Cumulocity GmbH
 
-
 from c8y_api import CumulocityApi
 from c8y_api.model import Operation
 
@@ -18,12 +17,12 @@ def test_CRUD(live_c8y: CumulocityApi, sample_device):
     operation = operation.create()
 
     # -> operation should have been created and in PENDING state
-    operations = live_c8y.operations.get_all(agent_id=sample_device.id, status=Operation.Status.PENDING)
+    operations = live_c8y.operations.get_all(device_id=sample_device.id, status=Operation.Status.PENDING)
     assert len(operations) == 1
     assert operations[0].id == operation.id
 
     # -> same result with get_last
-    operation2 = live_c8y.operations.get_last(agent_id=sample_device.id, status=Operation.Status.PENDING)
+    operation2 = live_c8y.operations.get_last(device_id=sample_device.id, status=Operation.Status.PENDING)
     assert operation2.id == operation.id
 
     # (2) update operation
@@ -45,3 +44,49 @@ def test_CRUD(live_c8y: CumulocityApi, sample_device):
 
     # -> cannot be found anymore
     assert not live_c8y.operations.get_all(device_id=sample_device.id)
+
+
+def test_get(live_c8y: CumulocityApi, sample_device):
+    """Verify that query-like retrieval works as expected."""
+    # (1) create operations
+    operations = [
+        Operation(live_c8y, sample_device.id, description=f'Description {i}',
+                  c8y_Command={'text': 'Command text'}).create()
+        for i in range(5)
+    ]
+
+    # (2) all should have been created an in PENDING state
+    result = live_c8y.operations.get_all(device_id=sample_device.id, status=Operation.Status.PENDING)
+    assert len(result) == 5
+    assert all(o.device_id == sample_device.id for o in result)
+
+    # (3) get last
+    result = live_c8y.operations.get_last(device_id=sample_device.id)
+    assert isinstance(result, Operation)
+    assert result.device_id == sample_device.id
+
+    # (4) retrieving subsets
+    operations[0].status = Operation.Status.EXECUTING
+    operations[1].status = Operation.Status.EXECUTING
+    operations[0].update()
+    operations[1].update()
+
+    result = live_c8y.operations.get_all(device_id=sample_device.id, status=Operation.Status.PENDING)
+    assert len(result) == 3
+    result = live_c8y.operations.get_last(device_id=sample_device.id, status=Operation.Status.PENDING)
+    assert result.status == Operation.Status.PENDING
+    assert result.device_id == sample_device.id
+
+    result = live_c8y.operations.get_all(device_id=sample_device.id, status=Operation.Status.EXECUTING)
+    assert len(result) == 2
+    result = live_c8y.operations.get_last(device_id=sample_device.id, status=Operation.Status.EXECUTING)
+    assert result.status == Operation.Status.EXECUTING
+    assert result.device_id == sample_device.id
+
+    # (5) deleting subsets
+    live_c8y.operations.delete_by(device_id=sample_device.id, status=Operation.Status.EXECUTING)
+    assert live_c8y.operations.get_all(device_id=sample_device.id, status=Operation.Status.EXECUTING) == []
+    assert len(live_c8y.operations.get_all(device_id=sample_device.id, status=Operation.Status.PENDING)) == 3
+
+    # (6) no match with get_last
+    assert live_c8y.operations.get_last(device_id=sample_device.id, status=Operation.Status.EXECUTING) is None

tasks.py

@@ -48,7 +48,7 @@ def init(c):
 def lint(c, scope='all'):
     """Run PyLint."""
     if scope == 'all':
-        scope = 'c8y_api c8y_tk tests integration_tests samples'
+        scope = 'c8y_api/ c8y_tk tests integration_tests samples'
     c.run(f'pylint --rcfile pylintrc --fail-under=9 {scope}')