Add Input/Config Validation Checks (#246)

* update pydocstyle convention

* named, array input validation

* checked namedinput requirements

* more input validation

python/reactivedataflow/pyproject.toml

@@ -130,7 +130,7 @@ ignore = [
 "*.ipynb" = ["T201"]
 
 [tool.ruff.lint.pydocstyle]
-convention = "numpy"
+convention = "google"
 
 [tool.pytest.ini_options]
 log_cli = true

python/reactivedataflow/reactivedataflow/bindings.py

@@ -11,11 +11,13 @@
 
 
 class ArrayInput(BaseModel, extra="allow"):
-    """Specification for an array-based input port."""
+    """Specification for an array-based input binding."""
 
-    type: str | None = Field(default=None, description="The type of the port.")
-    required: bool = Field(
-        default=False, description="Whether the input port is required."
+    type: str | None = Field(
+        default=None, description="The item-type of the array input port."
+    )
+    required: int | None = Field(
+        default=None, description="The minimum number of array inputs required."
     )
     parameter: str | None = Field(
         default=None,
@@ -81,16 +83,21 @@ class Config(BaseModel, extra="allow"):
 
 
 class Bindings:
-    """Node Binding Managemer class.
+    """Node Binding Manager class.
 
     Node bindings are used to map processing-graph inputs, outputs, and configuration values into the appropriate
     function parameters. This class is used to manage the bindings for a node.
     """
 
     _bindings: list[Binding]
 
-    def __init__(self, ports: list[Binding]):
-        self._bindings = ports
+    def __init__(self, bindings: list[Binding]):
+        """Initialize the Bindings object.
+
+        Args:
+            bindings: The list of bindings for the node.
+        """
+        self._bindings = bindings
         self._validate()
 
     def _validate(self):
@@ -111,24 +118,37 @@ def bindings(self) -> list[Binding]:
     @property
     def config(self) -> list[Config]:
         """Return the configuration bindings."""
-        return [port for port in self.bindings if isinstance(port, Config)]
+        return [b for b in self.bindings if isinstance(b, Config)]
 
     @property
     def input(self) -> list[Input]:
         """Return the input bindings."""
-        return [port for port in self.bindings if isinstance(port, Input)]
+        return [b for b in self.bindings if isinstance(b, Input)]
 
     @property
     def outputs(self) -> list[Output]:
         """Return the output bindings."""
-        return [port for port in self._bindings if isinstance(port, Output)]
+        return [b for b in self._bindings if isinstance(b, Output)]
 
     @property
     def array_input(self) -> ArrayInput | None:
-        """Return the array input port."""
+        """Return the array input binding."""
         return next((p for p in self._bindings if isinstance(p, ArrayInput)), None)
 
     @property
     def named_inputs(self) -> NamedInputs | None:
-        """Return the named inputs port."""
+        """Return the named inputs binding."""
         return next((p for p in self._bindings if isinstance(p, NamedInputs)), None)
+
+    @property
+    def required_input_names(self) -> set[str]:
+        """Return the required named inputs."""
+        result = {p.name for p in self.input if p.required}
+        if self.named_inputs:
+            result.update(self.named_inputs.required)
+        return result
+
+    @property
+    def required_config_names(self) -> set[str]:
+        """Return the required named inputs."""
+        return {p.name for p in self.config if p.required}

python/reactivedataflow/reactivedataflow/decorators/apply_decorators.py

@@ -10,9 +10,12 @@
 
 
 def apply_decorators(fn: AnyFn, decorators: list[Decorator]) -> AnyFn:
-    """
-    Apply a series of decorators to a function reference.
+    """Apply a series of decorators to a function reference.
 
     This is useful for splitting apart verb registration from the verb implementation.
+
+    Args:
+        fn: The function to decorate.
+        decorators: The decorators to apply. These will be applied in reverse order so that they can be copied/pasted from a decorated function.
     """
     return reduce(lambda x, y: y(x), reversed(decorators), fn)

python/reactivedataflow/reactivedataflow/decorators/connect_input.py

@@ -17,7 +17,7 @@ def connect_input(
     """Decorate an execution function with input conditions.
 
     Args:
-        required (list[str] | None): The list of required input names. Defaults to None. If present, the function will only execute if all required inputs are present.
+        bindings (Bindings): The input bindings for the function.
     """
 
     def wrap_fn(

python/reactivedataflow/reactivedataflow/decorators/connect_output.py

@@ -25,7 +25,8 @@ def connect_output(
     """Decorate an execution function with output conditions.
 
     Args:
-        default_output (bool): The default output of the function.
+        mode (OutputMode, optional): The output mode. Defaults to OutputMode.Value.
+        output_names (list[str] | None, optional): The output names. Defaults to None.
     """
     if mode == OutputMode.Tuple and output_names is None:
         raise OutputNamesMissingInTupleOutputModeError

python/reactivedataflow/reactivedataflow/errors.py

@@ -2,17 +2,43 @@
 """reactivedataflow Error Types."""
 
 
+class RequiredNodeInputNotFoundError(ValueError):
+    """An exception for required input not found."""
+
+    def __init__(self, nid: str, input_name: str):
+        """Initialize the RequiredNodeInputNotFoundError."""
+        super().__init__(f"Node {nid} is missing required input '{input_name}'.")
+
+
+class RequiredNodeArrayInputNotFoundError(ValueError):
+    """An exception for required array input not found."""
+
+    def __init__(self, nid: str):
+        """Initialize the RequiredNodeInputNotFoundError."""
+        super().__init__(f"Node {nid} is missing required array input.")
+
+
+class RequiredNodeConfigNotFoundError(ValueError):
+    """An exception for required input not found."""
+
+    def __init__(self, nid: str, config_key: str):
+        """Initialize the RequiredNodeInputNotFoundError."""
+        super().__init__(f"Node {nid} is missing required config '{config_key}'.")
+
+
 class NodeAlreadyDefinedError(ValueError):
     """An exception for adding a node that already exists."""
 
     def __init__(self, nid: str):
+        """Initialize the NodeAlreadyDefinedError."""
         super().__init__(f"Node {nid} already defined.")
 
 
 class NodeNotFoundError(KeyError):
     """An exception for unknown node."""
 
     def __init__(self, nid: str):
+        """Initialize the NodeNotFoundError."""
         super().__init__(f"Node {nid} not found.")
 
 
@@ -24,43 +50,57 @@ def __init__(self, name: str):
         super().__init__(f"Output '{name}' is already defined.")
 
 
+class InputNotFoundError(ValueError):
+    """An exception for input not defined."""
+
+    def __init__(self, input_name: str):
+        """Initialize the InputNotFoundError."""
+        super().__init__(f"Input '{input_name}' not found.")
+
+
 class OutputNotFoundError(ValueError):
     """An exception for output not defined."""
 
     def __init__(self, output_name: str):
+        """Initialize the OutputNotFoundError."""
         super().__init__(f"Output '{output_name}' not found.")
 
 
 class OutputNamesMissingInTupleOutputModeError(ValueError):
     """An exception for missing output names in tuple output mode."""
 
     def __init__(self):
+        """Initialize the OutputNamesMissingInTupleOutputModeError."""
         super().__init__("Output names are required in tuple output mode.")
 
 
 class OutputNamesNotValidInValueOutputModeError(ValueError):
     """An exception for output names in value output mode."""
 
     def __init__(self):
+        """Initialize the OutputNamesNotValidInValueOutputModeError."""
         super().__init__("Output names are not allowed in Value output mode")
 
 
 class VerbNotFoundError(KeyError):
     """An exception for unknown verb."""
 
     def __init__(self, name: str):
+        """Initialize the VerbNotFoundError."""
         super().__init__(f"Unknown verb: {name}")
 
 
 class VerbAlreadyDefinedError(ValueError):
     """An exception for already defined verb."""
 
     def __init__(self, name: str):
+        """Initialize the VerbAlreadyDefinedError."""
         super().__init__(f"Verb {name} already defined.")
 
 
 class PortNamesMustBeUniqueError(ValueError):
     """An exception for non-unique port names."""
 
     def __init__(self):
+        """Initialize the PortNamesMustBeUniqueError."""
         super().__init__("Port names must be unique.")

python/reactivedataflow/reactivedataflow/execution_graph.py

@@ -17,6 +17,12 @@ class ExecutionGraph:
     _outputs: dict[str, Output]
 
     def __init__(self, nodes: dict[str, Node], outputs: dict[str, Output]):
+        """Initialize the execution graph.
+
+        Args:
+            nodes: The nodes in the graph.
+            outputs: The outputs of the graph.
+        """
         self._nodes = nodes
         self._outputs = outputs
 
@@ -30,13 +36,13 @@ def output(self, name: str) -> rx.Observable[Any]:
         output = self._outputs.get(name)
         if output is None:
             raise OutputNotFoundError(name)
-        node = self._nodes[output.node]
+        node = self._nodes[output.node or output.name]
         return node.output(output.port)
 
     def output_value(self, name: str) -> rx.Observable[Any]:
         """Read the output of a node."""
         output = self._outputs.get(name)
         if output is None:
             raise OutputNotFoundError(name)
-        node = self._nodes[output.node]
+        node = self._nodes[output.node or output.name]
         return node.output_value(output.port)