Ability to parse expressions that can filter and/or map a list

generic_k8s_webhook/config_parser/entrypoint.py

@@ -110,6 +110,7 @@ def _parse_v1beta1(self, raw_list_webhook_config: dict) -> list[Webhook]:
                         op_parser.ForEachParser,
                         op_parser.MapParser,
                         op_parser.ContainParser,
+                        op_parser.FilterParser,
                         op_parser.ConstParser,
                         op_parser.GetValueParser,
                     ],

generic_k8s_webhook/config_parser/expr_parser.py

@@ -7,7 +7,16 @@
 from generic_k8s_webhook import utils
 
 GRAMMAR_V1 = r"""
-    ?start: expr
+    ?start: expr | list_filter_map
+
+    ?list_filter_map: reference filter_expr -> filterr
+        | reference map_expr -> mapp
+        | list_filter_map filter_expr -> filterr
+        | list_filter_map map_expr -> mapp
+
+    ?filter_expr: "|" expr
+
+    ?map_expr: "->" expr
 
     ?expr: or
 
@@ -33,12 +42,17 @@
         | product "*" atom  -> mul
         | product "/" atom  -> div
 
-    ?atom: SIGNED_NUMBER    -> number
-        | ESCAPED_STRING    -> const_string
-        | REF               -> ref
-        | BOOL              -> boolean
+    ?atom: signed_number
+        | escaped_string
+        | reference
+        | bool
         | "(" expr ")"
 
+    signed_number: SIGNED_NUMBER    -> number
+    escaped_string: ESCAPED_STRING  -> const_string
+    reference: REF                  -> ref
+    bool: BOOL                      -> boolean
+
     BOOL: "true" | "false"
     REF: "$"? ("."(CNAME|"*"|INT))+
 
@@ -113,6 +127,14 @@ def boolean(self, items):
         elem_bool = elem == "true"
         return op.Const(elem_bool)
 
+    def filterr(self, items):
+        elems, operator = items
+        return op.Filter(elems, operator)
+
+    def mapp(self, items):
+        elems, operator = items
+        return op.ForEach(elems, operator)
+
 
 def parse_ref(ref: str) -> op.GetValue:
     """Parses a string that is a reference to some element within a json payload
@@ -186,12 +208,13 @@ def get_transformer(cls) -> Transformer:
 def main():
     parser = Lark(GRAMMAR_V1)
     # print(parser.parse('.key != "some string"').pretty())
-    tree = parser.parse('"true" != "false"')
+    tree = parser.parse('.spec.containers | .name != "main" -> .requests.cpu * 0.75')
     print(tree.pretty())
     trans = MyTransformerV1()
     new_op = trans.transform(tree)
     print(new_op)
-    print(new_op.get_value([]))
+    context = {"spec": {"containers": [{"name": "main"}, {"name": "side", "requests": {"cpu": 2}}]}}
+    print(new_op.get_value([context]))
 
 
 if __name__ == "__main__":

generic_k8s_webhook/config_parser/operator_parser.py

@@ -239,6 +239,24 @@ def get_name(cls) -> str:
         return "map"
 
 
+class FilterParser(OperatorParser):
+    @classmethod
+    def get_name(cls) -> str:
+        return "filter"
+
+    def parse(self, op_inputs: dict | list, path_op: str) -> operators.Filter:
+        raw_elements = utils.must_get(op_inputs, "elements", f"In {path_op}, required 'elements'")
+        elements = self.meta_op_parser.parse(raw_elements, f"{path_op}.elements")
+
+        raw_op = utils.must_get(op_inputs, "op", f"In {path_op}, required 'op'")
+        op = self.meta_op_parser.parse(raw_op, f"{path_op}.op")
+
+        try:
+            return operators.Filter(elements, op)
+        except TypeError as e:
+            raise ParsingException(f"Error when parsing {path_op}") from e
+
+
 class ContainParser(OperatorParser):
     @classmethod
     def get_name(cls) -> str:

generic_k8s_webhook/operators.py

@@ -273,6 +273,30 @@ def return_type(self) -> type | None:
         return list[self.op.return_type()]
 
 
+class Filter(Operator):
+    def __init__(self, elements: Operator, op: Operator) -> None:
+        self.elements = elements
+        self.op = op
+
+    def get_value(self, contexts: list):
+        elements = self.elements.get_value(contexts)
+        if elements is None:
+            return []
+
+        result_list = []
+        for elem in elements:
+            mapped_elem = self.op.get_value(contexts + [elem])
+            if mapped_elem:
+                result_list.append(elem)
+        return result_list
+
+    def input_type(self) -> type | None:
+        return None
+
+    def return_type(self) -> type | None:
+        return list[self.op.return_type()]
+
+
 class Contain(Operator):
     def __init__(self, elements: Operator, elem: Operator) -> None:
         self.elements = elements

tests/conditions_test.yaml

@@ -236,6 +236,22 @@ test_suites:
                   - maxCPU: 1
                   - maxCPU: 2
             expected_result: false
+  - name: FILTER
+    tests:
+      - schemas: [v1beta1]
+        cases:
+          - condition:
+              filter:
+                elements:
+                  getValue: .containers
+                op: .maxCPU < 2
+            context:
+              - containers:
+                  - name: container1
+                    maxCPU: 1
+                  - name: container2
+                    maxCPU: 2
+            expected_result: [{ name: container1, maxCPU: 1 }]
   - name: RAW_STR_EXPR
     tests:
       - schemas: [v1beta1]
@@ -258,3 +274,27 @@ test_suites:
                   - maxCPU: 1
                   - maxCPU: 2
             expected_result: true
+  - name: LIST_FILTER_MAP_EXPR
+    tests:
+      - schemas: [v1beta1]
+        cases:
+          - condition: .containers | .name != "main"
+            context:
+              - containers:
+                  - name: main
+                  - name: istio
+            expected_result: [name: istio]
+          - condition: ".containers -> .maxCPU * 2"
+            context:
+              - containers:
+                  - maxCPU: 1
+                  - maxCPU: 2
+            expected_result: [2, 4]
+          - condition: .containers | .name != "main" -> .maxCPU > 1
+            context:
+              - containers:
+                  - name: main
+                    maxCPU: 1
+                  - name: istio
+                    maxCPU: 2
+            expected_result: [true]