added count method

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 2.1.x
+### 2.1.0
+#### Features
+* Added count method in Routing
+
 ## 2.0.x
 ### 2.0.0
 #### Changes

routing_test.py

@@ -456,6 +456,17 @@ def test_get_stats(self):
         self.assertDictEqual(self.routing.get_stats(delete=True), expected_stats)
         self.assertDictEqual(self.routing.get_stats(), {"streams": {"exists-fh0wery": {}, "equals-fbh49ry29": {}}, "customer": {"customer-dh8rh9fow": {}}})
 
+    def test_count(self):
+        rule_list = [
+            load_test_data("test_rule_1_equals"),
+            load_test_data("test_rule_3_customer_equals"),
+            load_test_data("test_rule_4_multiple_filters"),
+            load_test_data("test_rule_5_exists"),
+            load_test_data("test_rule_6_not_exists"),
+        ]
+        self.routing.load_from_dicts(rule_list)
+        self.assertEqual(self.routing.count(), 5)
+
 
 if __name__ == "__main__":
     unittest.main()

routingfilter/filters/rule.py

@@ -100,6 +100,15 @@ def __init__(self, tag: str):
         self._rules = []
         self.logger = logging.getLogger(self.__class__.__name__)
 
+    def count(self) -> int:
+        """
+        Return the number of rules.
+
+        :return: number of rules
+        :rtype: int
+        """
+        return len(self._rules)
+
     def match(self, event: DictQuery, tag: str) -> Results | None:
         """
         Call all match methods of the Rules and return the result of first match.

routingfilter/filters/stream.py

@@ -13,6 +13,18 @@ def __init__(self, stream):
         self._ruleManagers = {}
         self.logger = logging.getLogger(self.__class__.__name__)
 
+    def count(self) -> int:
+        """
+        Returns the number of rules in the stream.
+
+        :return: number of rules
+        :rtype: int
+        """
+        count_rules = 0
+        for key in self._ruleManagers.keys():
+            count_rules += self._ruleManagers[key].count()
+        return count_rules
+
     def match(self, event: DictQuery, tag_field_name: str) -> List[Results]:
         """
         Call all ruleManagers that contain tha tag of event "tags" field (that could be a list).

routingfilter/routing.py

@@ -17,6 +17,17 @@ def __init__(self):
         self.variables = {}
         self.logger = logging.getLogger(self.__class__.__name__)
 
+    def count(self) -> int:
+        """
+        Return the number of the rules.
+
+        :return: number of the rules
+        :rtype: int
+        """
+        streams_count = self.streams.count()
+        customer_count = self.customer.count()
+        return streams_count + customer_count
+
     def get_stats(self, delete: bool = False) -> dict:
         """
         Call get_stats of streams and return the stats. If delete is True, reset the stats.

setup.py

@@ -5,7 +5,7 @@
 
 setup(
     name="routingfilter",
-    version="2.0.0",
+    version="2.1.0",
     packages=["routingfilter"],
     include_package_data=True,
     install_requires=["IPy~=1.1", "macaddress~=2.0.2"],