Provide helpers to combine few callbacks for body-/meta-filters

docs/filters.rst

@@ -83,3 +83,32 @@ By arbitrary callbacks
                     annotations={'some-annotation': check_value})
     def my_handler(spec, **_):
         pass
+
+Kopf provides few helpers to combine multiple callbacks into one
+(the semantics is the same as for Python's built-in functions)::
+
+    import kopf
+
+    def whole_fn1(name, **_): return name.startswith('kopf-')
+    def whole_fn2(spec, **_): return spec.get('field') == 'value'
+    def value_fn1(value, **_): return value.startswith('some')
+    def value_fn2(value, **_): return value.endswith('label')
+
+    @kopf.on.create('zalando.org', 'v1', 'kopfexamples',
+                    when=kopf.all_([whole_fn1, whole_fn2]),
+                    labels={'somelabel': kopf.all_([value_fn1, value_fn2])})
+    def create_fn1(**_):
+        pass
+
+    @kopf.on.create('zalando.org', 'v1', 'kopfexamples',
+                    when=kopf.any_([whole_fn1, whole_fn2]),
+                    labels={'somelabel': kopf.any_([value_fn1, value_fn2])})
+    def create_fn2(**_):
+        pass
+
+The following wrappers are available:
+
+* `kopf.not_(fn)` -- the function must return ``False`` to pass the filters.
+* `kopf.any_([...])` -- at least one of the functions must return ``True``.
+* `kopf.all_([...])` -- all of the functions must return ``True``.
+* `kopf.none_([...])` -- all of the functions must return ``False``.

kopf/__init__.py

@@ -94,6 +94,12 @@
     build_object_reference,
     build_owner_reference,
 )
+from kopf.structs.callbacks import (
+    not_,
+    all_,
+    any_,
+    none_,
+)
 from kopf.structs.configuration import (
     OperatorSettings,
 )
@@ -162,6 +168,10 @@
     'event', 'info', 'warn', 'exception',
     'spawn_tasks', 'run_tasks', 'operator', 'run', 'create_tasks',
     'adopt', 'label',
+    'not_',
+    'all_',
+    'any_',
+    'none_',
     'get_default_lifecycle', 'set_default_lifecycle',
     'build_object_reference', 'build_owner_reference',
     'append_owner_reference', 'remove_owner_reference',

kopf/structs/callbacks.py

@@ -5,7 +5,7 @@
 not so important for the codebase, they are moved to this separate module.
 """
 import logging
-from typing import NewType, Any, Union, Optional, Callable, Coroutine
+from typing import NewType, Any, Collection, Union, Optional, Callable, Coroutine, TypeVar
 
 from typing_extensions import Protocol
 
@@ -171,3 +171,30 @@ def __call__(  # lgtm[py/similar-function]
             logger: Union[logging.Logger, logging.LoggerAdapter],
             **kwargs: Any,
     ) -> bool: ...
+
+
+_FnT = TypeVar('_FnT', WhenFilterFn, MetaFilterFn)
+
+
+def not_(fn: _FnT) -> _FnT:
+    def not_fn(*args: Any, **kwargs: Any) -> bool:
+        return not fn(*args, **kwargs)
+    return not_fn
+
+
+def all_(fns: Collection[_FnT]) -> _FnT:
+    def all_fn(*args: Any, **kwargs: Any) -> bool:
+        return all(fn(*args, **kwargs) for fn in fns)
+    return all_fn
+
+
+def any_(fns: Collection[_FnT]) -> _FnT:
+    def any_fn(*args: Any, **kwargs: Any) -> bool:
+        return any(fn(*args, **kwargs) for fn in fns)
+    return any_fn
+
+
+def none_(fns: Collection[_FnT]) -> _FnT:
+    def none_fn(*args: Any, **kwargs: Any) -> bool:
+        return not any(fn(*args, **kwargs) for fn in fns)
+    return none_fn

tests/test_filtering_helpers.py

@@ -0,0 +1,101 @@
+import kopf
+
+
+def _never1(*_, **__):
+    return False
+
+
+def _never2(*_, **__):
+    return False
+
+
+def _always1(*_, **__):
+    return True
+
+
+def _always2(*_, **__):
+    return True
+
+
+def test_notfn_when_true():
+    combined = kopf.not_(_always1)
+    result = combined()
+    assert result is False
+
+
+def test_notfn_when_false():
+    combined = kopf.not_(_never1)
+    result = combined()
+    assert result is True
+
+
+def test_allfn_when_all_are_true():
+    combined = kopf.all_([_always1, _always2])
+    result = combined()
+    assert result is True
+
+
+def test_allfn_when_one_is_false():
+    combined = kopf.all_([_always1, _never1])
+    result = combined()
+    assert result is False
+
+
+def test_allfn_when_all_are_false():
+    combined = kopf.all_([_never1, _never2])
+    result = combined()
+    assert result is False
+
+
+def test_allfn_when_no_functions():
+    combined = kopf.all_([])
+    result = combined()
+    assert result is True
+
+
+def test_anyfn_when_all_are_true():
+    combined = kopf.any_([_always1, _always2])
+    result = combined()
+    assert result is True
+
+
+def test_anyfn_when_one_is_false():
+    combined = kopf.any_([_always1, _never1])
+    result = combined()
+    assert result is True
+
+
+def test_anyfn_when_all_are_false():
+    combined = kopf.any_([_never1, _never2])
+    result = combined()
+    assert result is False
+
+
+def test_anyfn_when_no_functions():
+    combined = kopf.any_([])
+    result = combined()
+    assert result is False
+
+
+def test_nonefn_when_all_are_true():
+    combined = kopf.none_([_always1, _always2])
+    result = combined()
+    assert result is False
+
+
+def test_nonefn_when_one_is_false():
+    combined = kopf.none_([_always1, _never1])
+    result = combined()
+    assert result is False
+
+
+def test_nonefn_when_all_are_false():
+    combined = kopf.none_([_never1, _never2])
+    result = combined()
+    assert result is True
+
+
+def test_nonefn_when_no_functions():
+    combined = kopf.none_([])
+    result = combined()
+    assert result is True