Merge pull request #121 from AllenNeuralDynamics:feat-run-if

Add `run_if` decorator to skip execution given a predicate

Author: bruno-f-cruz

src/clabe/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.6.8"
+__version__ = "0.6.9"
 
 import logging
 

src/clabe/launcher/_callable_manager.py

@@ -194,3 +194,30 @@ def wrapper(*args, **kwargs):
         return wrapper
 
     return decorator
+
+
+def run_if(predicate: t.Callable[..., bool]) -> t.Callable[[t.Callable[P, R]], t.Callable[P, Optional[R]]]:
+    """
+    A decorator that only runs the wrapped function if the predicate returns True.
+    If the predicate returns False, returns None.
+
+    Args:
+        predicate: A callable that takes the same arguments as the wrapped function and returns a boolean.
+
+    Returns:
+        The decorated function that runs only if predicate(*args, **kwargs) is True, else returns None.
+    """
+
+    def decorator(func: t.Callable[P, R]) -> t.Callable[P, Optional[R]]:
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            fn_name = getattr(func, "__name__", repr(func))
+            if predicate(*args, **kwargs):
+                logger.debug(f"Predicate passed for {fn_name}, executing function")
+                return func(*args, **kwargs)
+            logger.debug(f"Predicate failed for {fn_name}, skipping execution")
+            return None
+
+        return wrapper
+
+    return decorator

tests/launcher/test_callable_manager.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from clabe.launcher._callable_manager import Promise, _CallableManager, _UnsetType, ignore_errors
+from clabe.launcher._callable_manager import Promise, _CallableManager, _UnsetType, ignore_errors, run_if
 
 
 class TestCallableManager:
@@ -278,3 +278,49 @@ def test_ignore_errors_lambda_function(self, caplog):
         assert result2 == "lambda_failed"
         # Lambda functions have a generic name
         assert "Exception in <lambda>: division by zero" in caplog.text
+
+
+class TestRunIfDecorator:
+    def test_run_if_runs_when_predicate_true(self):
+        def always_true(*args, **kwargs):
+            return True
+
+        @run_if(always_true)
+        def my_func(x):
+            return x * 2
+
+        assert my_func(3) == 6
+
+    def test_run_if_returns_none_when_predicate_false(self):
+        def always_false(*args, **kwargs):
+            return False
+
+        @run_if(always_false)
+        def my_func(x):
+            return x * 2
+
+        assert my_func(3) is None
+
+    def test_run_if_predicate_depends_on_args(self):
+        def is_positive(x):
+            return x > 0
+
+        @run_if(is_positive)
+        def square(x):
+            return x * x
+
+        assert square(2) == 4
+        assert square(-1) is None
+
+    def test_run_if_preserves_function_metadata(self):
+        def always_true(*args, **kwargs):
+            return True
+
+        @run_if(always_true)
+        def documented_func(x):
+            """This function squares its input."""
+            return x * x
+
+        assert hasattr(documented_func, "__name__")
+        assert hasattr(documented_func, "__doc__")
+        assert documented_func.__doc__ == "This function squares its input."