Add complexity check with complexipy

Author: goanpeca

.pre-commit-config.yaml

@@ -24,3 +24,9 @@ repos:
       # Run the formatter
       - id: ruff-format
         types_or: [python, pyi]
+
+  # See https://github.com/rohaquinlop/complexipy
+  - repo: https://github.com/rohaquinlop/complexipy-pre-commit
+    rev: v3.0.0 # Use the latest version
+    hooks:
+      - id: complexipy

datalayer_core/sdk/decorators.py

@@ -13,6 +13,53 @@
 # - incorrect number of args
 
 
+def _process_args_kwargs_mapping(
+    func: Callable[..., Any],
+    args: tuple,
+    kwargs: dict,
+    inputs: list[str],
+    variables: dict[str, Any],
+) -> dict[str, Any]:
+    sig = inspect.signature(func)
+    mapping = {}
+
+    for idx, (name, _param) in enumerate(sig.parameters.items()):
+        mapping[name] = inputs[idx]
+
+    for kwarg, kwarg_value in kwargs.items():
+        variables[mapping[kwarg]] = kwarg_value
+
+    for idx, (arg_value) in enumerate(args):
+        kwarg = inputs[idx]
+        variables[kwarg] = arg_value
+
+    return variables
+
+
+def _process_inputs(
+    func: Callable[..., Any],
+    inputs_decorated: Optional[list[str]],
+    variables: dict[str, Any],
+) -> tuple[dict[str, Any], list[str]]:
+    sig = inspect.signature(func)
+    if inputs_decorated is None:
+        inputs = []
+        for name, _param in sig.parameters.items():
+            inputs.append(name)
+            variables[name] = (
+                _param.default
+                if _param.default is not inspect.Parameter.empty
+                else None
+            )
+    else:
+        if len(sig.parameters) != len(inputs_decorated):
+            raise ValueError(
+                f"Function {func.__name__} has {len(sig.parameters)} parameters, "
+                f"but {len(inputs_decorated)} inputs were provided."
+            )
+    return variables, inputs
+
+
 def datalayer(
     runtime_name: Union[Callable[..., Any], Optional[str]] = None,
     inputs: Optional[list[str]] = None,
@@ -65,37 +112,20 @@ def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
         if output_decorated is None:
             output = f"DATALAYER_RUNTIME_OUTPUT_{func.__name__}".upper()
 
-        sig = inspect.signature(func)
-        if inputs_decorated is None:
-            inputs = []
-            for name, _param in sig.parameters.items():
-                inputs.append(name)
-                variables[name] = (
-                    _param.default
-                    if _param.default is not inspect.Parameter.empty
-                    else None
-                )
-        else:
-            if len(sig.parameters) != len(inputs_decorated):
-                raise ValueError(
-                    f"Function {func.__name__} has {len(sig.parameters)} parameters, "
-                    f"but {len(inputs_decorated)} inputs were provided."
-                )
+        variables, inputs = _process_inputs(
+            func=func, inputs_decorated=inputs_decorated, variables=variables
+        )
 
         @functools.wraps(func)
         def wrapper(*args: Any, **kwargs: Any) -> Any:
-            sig = inspect.signature(func)
-            mapping = {}
-            for idx, (name, _param) in enumerate(sig.parameters.items()):
-                mapping[name] = (inputs_decorated or inputs)[idx]
-
-            for kwarg, kwarg_value in kwargs.items():
-                variables[mapping[kwarg]] = kwarg_value
-
-            for idx, (arg_value) in enumerate(args):
-                kwarg = (inputs_decorated or inputs)[idx]
-                variables[kwarg] = arg_value
-
+            # Process args and kwargs to map to inputs
+            variables = _process_args_kwargs_mapping(
+                func=func,
+                args=args,
+                kwargs=kwargs,
+                inputs=(inputs_decorated or inputs),
+                variables=variables,
+            )
             function_call = (
                 f"{output_decorated or output} = {func.__name__}("
                 + ", ".join(inputs_decorated or inputs)