Feature: client side filtering

CHANGELOG.md

@@ -1,10 +1,17 @@
 # Changelog
 
- * Updated and rewrote Notification 2.0 listener implementation. Added additional parameters for more control: 
-   `consumer_name`, `shared`, `auto_ack` and `auto_unsubscribe`. Added `unsubscribe` function for removing
-   subscribers on demand. Both `AsyncListener` and `Listener` now provide consistent `start`/`stop` functions
-   which take care of coroutine and thread creation. The `listen ` function can still be invoked directly if
-   necessary. 
+* Added client-side filtering to many of the standard API (wherever sensibly applicable); These APIs `select`
+  and `get_all` functions now feature optional `include` and `exclude` parameters which can be used to filter 
+  the results before being wrapped into Python objects; Added multiple matchers including a JSONPath matcher
+  and a JMESPath matcher.
+* Added `QueueListener` and `AsyncQueueListener` classes to the Notification 2.0 toolkit. These pre-defined
+  listener implementation append new notifications to standard queues that can be monitored/listened to which 
+  makes Notification 2.0 solutions even simpler to implement.
+* Updated and rewrote Notification 2.0 listener implementation. Added additional parameters for more control: 
+  `consumer_name`, `shared`, `auto_ack` and `auto_unsubscribe`. Added `unsubscribe` function for removing
+  subscribers on demand. Both `AsyncListener` and `Listener` now provide consistent `start`/`stop` functions
+  which take care of coroutine and thread creation. The `listen ` function can still be invoked directly if
+  necessary. 
 
 ## Version 3.3.0
 

buster37.dockerfile

@@ -1,6 +1,10 @@
 FROM python:3.7-slim-buster
 
-RUN apt-get update && apt-get -y install git
+RUN sed -i 's|deb.debian.org/debian|archive.debian.org/debian|g' /etc/apt/sources.list \
+ && sed -i 's|security.debian.org/debian-security|archive.debian.org/debian-security|g' /etc/apt/sources.list \
+ && apt-get update \
+ && apt-get -y install git \
+ && rm -rf /var/lib/apt/lists/*
 
 COPY requirements.txt /
 RUN pip install --upgrade pip && pip install -r requirements.txt

c8y_api/model/_base.py

@@ -10,7 +10,12 @@
 from deprecated import deprecated
 
 from c8y_api._base_api import CumulocityRestApi
+from c8y_api.model.matcher import JsonMatcher
 from c8y_api.model._util import _DateUtil, _StringUtil, _QueryUtil
+try:
+    from c8y_api.model.matcher import JmesPathMatcher
+except ImportError:
+    pass
 
 
 def get_by_path(dictionary: dict, path: str, default: Any = None) -> Any:
@@ -633,6 +638,8 @@ def __init__(self, c8y: CumulocityRestApi, resource: str):
         # the default object name would be the resource path element just before
         # the last event for e.g. /event/events
         self.object_name = self.resource.split('/')[-1]
+        # the default JSON matcher for client-side filtering
+        self.default_matcher = JmesPathMatcher
 
     def build_object_path(self, object_id: int | str) -> str:
         """Build the path to a specific object of this resource.
@@ -777,16 +784,33 @@ def _get_count(self, base_query: str) -> int:
         result_json = self.c8y.get(base_query + '&pageSize=1&withTotalPages=true')
         return result_json['statistics']['totalPages']
 
-    def _iterate(self, base_query: str, page_number: int | None, limit: int | None, parse_fun):
+    def _iterate(
+            self,
+            base_query: str,
+            page_number: int | None,
+            limit: int | None,
+            include: str | JsonMatcher | None,
+            exclude: str | JsonMatcher | None,
+            parse_fun
+    ):
         # if no specific page is defined we just start at 1
         current_page = page_number if page_number else 1
         # we will read page after page until
         #  - we reached the limit, or
         #  - there is no result (i.e. we were at the last page)
         num_results = 0
+        # compile/prepare filter if defined
+        if isinstance(include, str):
+            include = self.default_matcher(include)
+        if isinstance(exclude, str):
+            exclude = self.default_matcher(exclude)
+
         while True:
-            results = [parse_fun(x) for x in self._get_page(base_query, current_page)]
-            # no results, so we are done
+            results = [
+                parse_fun(x) for x in self._get_page(base_query, current_page)
+                if (not include or include.safe_matches(x))
+                   and (not exclude or not exclude.safe_matches(x))
+            ]
             if not results:
                 break
             for result in results:

c8y_api/model/_util.py

@@ -5,7 +5,6 @@
 from typing import Union
 
 from dateutil import parser
-from re import sub
 
 
 class _StringUtil(object):
@@ -30,6 +29,26 @@ def to_pascal_case(name: str):
             return name
         return parts[0] + "".join([x.title() for x in parts[1:]])
 
+    @staticmethod
+    def like(expression: str, string: str):
+        """Check if like-expression matches a string.
+
+        Only supports * at beginning and end.
+        """
+        return (
+            expression[1:-1] in string if expression.startswith('*') and expression.endswith('*')
+            else string.startswith(expression[:-1]) if expression.endswith('*')
+            else string.endswith(expression[1:]) if expression.startswith('*')
+            else expression == string
+        )
+
+    @staticmethod
+    def matches(expression: str, string: str):
+        """Check if regex expression matches a string."""
+        try:
+            return re.search(expression, string) is not None
+        except re.error:
+            return False
 
 class _QueryUtil(object):
 
@@ -39,15 +58,15 @@ def encode_odata_query_value(value):
         http://docs.oasis-open.org/odata/odata/v4.01/odata-v4.01-part2-url-conventions.html#sec_URLParsing
         http://docs.oasis-open.org/odata/odata/v4.01/cs01/abnf/odata-abnf-construction-rules.txt """
         # single quotes escaped through single quote
-        return sub('\'', '\'\'', value)
+        return re.sub('\'', '\'\'', value)
 
     @staticmethod
     def encode_odata_text_value(value):
         """Encode value strings according to OData query rules.
         http://docs.oasis-open.org/odata/odata/v4.01/odata-v4.01-part2-url-conventions.html#sec_URLParsing
         http://docs.oasis-open.org/odata/odata/v4.01/cs01/abnf/odata-abnf-construction-rules.txt """
         # single quotes escaped through single quote
-        encoded_quotes = sub('\'', '\'\'', value)
+        encoded_quotes = re.sub('\'', '\'\'', value)
         return encoded_quotes if " " not in encoded_quotes else f"'{encoded_quotes}'"
 
 

c8y_api/model/administration.py

@@ -11,6 +11,7 @@
 from c8y_api.model._base import CumulocityResource, SimpleObject
 from c8y_api.model._parser import SimpleObjectParser, ComplexObjectParser, as_values as parse_as_values
 from c8y_api.model._util import _DateUtil
+from c8y_api.model.matcher import JsonMatcher
 
 
 class PermissionUtil:
@@ -261,9 +262,9 @@ def from_json(cls, json: dict) -> GlobalRole:
         role: GlobalRole = cls._from_json(json, GlobalRole())
         # role ID are int for some reason - convert for consistency
         role.id = str(role.id)
-        if json['roles'] and json['roles']['references']:
+        if 'roles' in json and json['roles'] and json['roles']['references']:
             role.permission_ids = {ref['role']['id'] for ref in json['roles']['references']}
-        if json['applications']:
+        if 'applications' in json and json['applications']:
             role.application_ids = {ref['id'] for ref in json['applications']}
         return role
 
@@ -939,7 +940,14 @@ def get(self, role_id: str | int) -> InventoryRole:
         role.c8y = self.c8y  # inject c8y connection into instance
         return role
 
-    def select(self, limit: int = None, page_size: int = 1000, page_number: int = None) -> Generator[InventoryRole]:
+    def select(
+            self,
+            limit: int = None,
+            include: str | JsonMatcher = None,
+            exclude: str | JsonMatcher = None,
+            page_size: int = 1000,
+            page_number: int = None,
+    ) -> Generator[InventoryRole]:
         """Get all defined inventory roles.
 
         This function is implemented in a lazy fashion - results will only be
@@ -949,6 +957,12 @@ def select(self, limit: int = None, page_size: int = 1000, page_number: int = No
 
         Args:
             limit (int): Limit the number of results to this number.
+            include (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The inclusion is applied first.
+                Creates a JMESPath matcher by default for strings.
+            exclude (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The exclusion is applied second.
+                Creates a JMESPath matcher by default for strings.
             page_size (int): Define the number of objects read (and parsed
                 in one chunk). This is a performance related setting.
             page_number (int): Pull a specific page; this effectively disables
@@ -958,9 +972,15 @@ def select(self, limit: int = None, page_size: int = 1000, page_number: int = No
             Generator for InventoryRole objects
         """
         base_query = self._prepare_query(page_size=page_size)
-        return super()._iterate(base_query, page_number, limit, InventoryRole.from_json)
+        return super()._iterate(base_query, page_number, limit, include, exclude, InventoryRole.from_json)
 
-    def get_all(self, limit: int = None, page_size: int = 1000, page_number: int = None) -> List[InventoryRole]:
+    def get_all(
+            self,
+            limit: int = None,
+            include: str | JsonMatcher = None, exclude: str | JsonMatcher = None,
+            page_size: int = 1000,
+            page_number: int = None,
+    ) -> List[InventoryRole]:
         """Get all defined inventory roles.
 
         This function is a greedy version of the ``select`` function. All
@@ -971,7 +991,12 @@ def get_all(self, limit: int = None, page_size: int = 1000, page_number: int = N
         Returns:
             List of InventoryRole objects
         """
-        return list(self.select(limit=limit, page_size=page_size, page_number=page_number))
+        return list(self.select(
+            limit=limit,
+            include=include,
+            exclude=exclude,
+            page_size=page_size,
+            page_number=page_number))
 
     def select_assignments(self, username: str) -> Generator[InventoryRoleAssignment]:
         """Get all inventory role assignments of a user.
@@ -1065,6 +1090,7 @@ def select(
             only_devices: bool = None,
             with_subusers_count: bool = None,
             limit: int = None,
+            include: str | JsonMatcher = None, exclude: str | JsonMatcher = None,
             page_size: int = 5,
             page_number: int = None,
             as_values: str | tuple | list[str | tuple] = None,
@@ -1088,6 +1114,12 @@ def select(
             with_subusers_count (bool):  Whether to include an additional field
              `subusersCount` which holds the number of direct sub users.
             limit (int): Limit the number of results to this number.
+            include (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The inclusion is applied first.
+                Creates a JMESPath matcher by default for strings.
+            exclude (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The exclusion is applied second.
+                Creates a JMESPath matcher by default for strings.
             page_size (int): Define the number of events which are read (and
                 parsed in one chunk). This is a performance related setting.
             page_number (int): Pull a specific page; this effectively disables
@@ -1131,6 +1163,8 @@ def select(
             base_query,
             page_number,
             limit,
+            include,
+            exclude,
             User.from_json if not as_values else
             lambda x: parse_as_values(x, as_values))
 
@@ -1141,6 +1175,7 @@ def get_all(
             owner: str = None,
             only_devices: bool = None,
             with_subusers_count: bool = None,
+            include: str | JsonMatcher = None, exclude: str | JsonMatcher = None,
             page_size: int = 1000,
             as_values: str | tuple | list[str|tuple] = None,
             **kwargs
@@ -1160,6 +1195,8 @@ def get_all(
             owner=owner,
             only_devices=only_devices,
             with_subusers_count=with_subusers_count,
+            include=include,
+            exclude=exclude,
             page_size=page_size,
             as_values=as_values,
             **kwargs))
@@ -1286,12 +1323,24 @@ def get(self, role_id: int | str) -> GlobalRole:
                 self._global_roles_by_name = {g.name: g for g in self.get_all()}
             return self._global_roles_by_name[role_id]
 
-    def select(self, username: str = None, page_size: int = 5) -> Generator[GlobalRole]:
+    def select(
+            self,
+            username: str = None,
+            include: str | JsonMatcher = None,
+            exclude: str | JsonMatcher = None,
+            page_size: int = 5,
+    ) -> Generator[GlobalRole]:
         """Iterate over global roles.
 
         Args:
             username (str): Retrieve global roles assigned to a specified user
                 If omitted, all available global roles are returned
+            include (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The inclusion is applied first.
+                Creates a JMESPath matcher by default for strings.
+            exclude (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The exclusion is applied second.
+                Creates a JMESPath matcher by default for strings.
             page_size (int): Maximum number of entries fetched per requests;
                 this is a performance setting
 
@@ -1301,12 +1350,24 @@ def select(self, username: str = None, page_size: int = 5) -> Generator[GlobalRo
         # unfortunately, as selecting by username can't be implemented using the
         # generic _iterate method, we have to do everything manually.
         if username:
+            # compile/prepare filter if defined
+            if isinstance(include, str):
+                include = self.default_matcher(include)
+            if isinstance(exclude, str):
+                exclude = self.default_matcher(exclude)
             # select by username
             query = f'/user/{self.c8y.tenant_id}/users/{username}/groups?pageSize={page_size}&currentPage='
             page_number = 1
             while True:
                 response_json = self.c8y.get(query + str(page_number))
-                references = response_json['references']
+                references = (
+                    response_json['references'] if not include and not exclude
+                    else [
+                        x for x in response_json['references']
+                        if (not include or include.safe_matches(x['group']))
+                           and (not exclude or not exclude.safe_matches(x['group']))
+                    ]
+                )
                 if not references:
                     break
                 for ref_json in references:
@@ -1317,21 +1378,39 @@ def select(self, username: str = None, page_size: int = 5) -> Generator[GlobalRo
         else:
             # select all
             base_query = self._prepare_query(page_size=page_size)
-            yield from super()._iterate(base_query, page_number=None, limit=None, parse_fun=GlobalRole.from_json)
+            yield from super()._iterate(
+                base_query,
+                page_number=None,
+                limit=None,
+                include=include,
+                exclude=exclude,
+                parse_fun=GlobalRole.from_json)
 
-    def get_all(self, username: str = None, page_size: int = 1000) -> List[GlobalRole]:
+    def get_all(
+            self,
+            username: str = None,
+            include: str | JsonMatcher = None,
+            exclude: str | JsonMatcher = None,
+            page_size: int = 1000
+    ) -> List[GlobalRole]:
         """Retrieve global roles.
 
         Args:
             username (str): Retrieve global roles assigned to a specified user
                 If omitted, all available global roles are returned
+            include (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The inclusion is applied first.
+                Creates a JMESPath matcher by default for strings.
+            exclude (str | JsonMatcher): Matcher/expression to filter the query
+                results (on client side). The exclusion is applied second.
+                Creates a JMESPath matcher by default for strings.
             page_size (int): Maximum number of entries fetched per requests;
                 this is a performance setting
 
         Return:
             List of GlobalRole instances
         """
-        return list(self.select(username=username, page_size=page_size))
+        return list(self.select(username=username, include=include, exclude=exclude, page_size=page_size))
 
     def assign_users(self, role_id: int | str, *usernames: str):
         """Add users to a global role.