df: input flow: Add get_alternate_definitions helper function

We had a lack of clarity around what was going on with alternate
definitions in commit 8657032. There
were several if statements where we are looping through the origins
and if the origin is a dict, we check if the value is a tuple or a list.

If it is not, the key is the operation instance name and the value is
the output name for that operation instance.

If it is, the key is the origin and the value is the list or tuple of
definition names which are acceptable alternate definitions when the
input is coming from that origin.

This patch adds a helper method to make that more clear.

Signed-off-by: John Andersen <[email protected]>

Author: pdxjohnny

dffml/df/memory.py

@@ -490,12 +490,10 @@ async def gather_inputs(
                     # Ensure all conditions from all origins are True
                     for origin in origins:
                         # See comment in input_flow.inputs section
-                        alternate_definitions = []
-                        if isinstance(origin, tuple) and isinstance(
-                            origin[1], (list, tuple)
-                        ):
-                            alternate_definitions = origin[1]
-                            origin = origin[0]
+                        (
+                            alternate_definitions,
+                            origin,
+                        ) = input_flow.get_alternate_definitions(origin)
                         # Bail if the condition doesn't exist
                         if not origin in by_origin:
                             return
@@ -547,12 +545,10 @@ async def gather_inputs(
                             # These definitions will be used instead of the
                             # default one the input specified for the
                             # operation).
-                            alternate_definitions = []
-                            if isinstance(origin, tuple) and isinstance(
-                                origin[1], (list, tuple)
-                            ):
-                                alternate_definitions = origin[1]
-                                origin = origin[0]
+                            (
+                                alternate_definitions,
+                                origin,
+                            ) = input_flow.get_alternate_definitions(origin)
                             # Don't try to grab inputs from an origin that
                             # doesn't have any to give us
                             if not origin in by_origin:

dffml/df/types.py

@@ -429,6 +429,41 @@ def export(self):
     def _fromdict(cls, **kwargs):
         return cls(**kwargs)
 
+    @staticmethod
+    def get_alternate_definitions(
+        origin: Tuple[Union[List[str], Tuple[str]], str]
+    ) -> Tuple[Union[List[str], Tuple[str]], str]:
+        """
+        Returns the alternate definitions and origin for an entry within an input
+        flow. If there are no alternate defintions then the first element of the
+        returned tuple is an empty list.
+
+        Examples
+        --------
+
+        >>> from dffml import InputFlow
+        >>>
+        >>> input_flow = InputFlow(
+        ...     inputs={
+        ...         "features": [
+        ...             {"seed": ["Years", "Expertise", "Trust", "Salary"]}
+        ...         ],
+        ...         "token": [
+        ...             "client",
+        ...         ]
+        ...     }
+        ... )
+        >>>
+        >>> input_flow.get_alternate_definitions(list(input_flow.inputs["features"][0].items())[0])
+        (['Years', 'Expertise', 'Trust', 'Salary'], 'seed')
+        >>>
+        input_flow.get_alternate_definitions(list(input_flow.inputs["other"][0].items())[0])
+        ([], 'client')
+        """
+        if isinstance(origin, tuple) and isinstance(origin[1], (list, tuple)):
+            return origin[1], origin[0]
+        return [], origin
+
 
 @dataclass
 class Forward:
@@ -600,8 +635,9 @@ def update_by_origin(self):
                     )
                 else:
                     for origin in output_source.items():
-                        if isinstance(origin[1], (list, tuple,)):
-                            origin = origin[0]
+                        _, origin = input_flow.get_alternate_definitions(
+                            origin
+                        )
                         self.by_origin[operation.stage].setdefault(origin, [])
                         self.by_origin[operation.stage][origin].append(
                             operation
@@ -623,8 +659,9 @@ def update_by_origin(self):
                             # the Input.origin (like "seed"). And the value
                             # (origin[1]) is the list of definitions which are
                             # acceptable from that origin for this input.
-                            if isinstance(origin[1], (list, tuple,)):
-                                origin = origin[0]
+                            _, origin = input_flow.get_alternate_definitions(
+                                origin
+                            )
                             self.by_origin[operation.stage].setdefault(
                                 origin, []
                             )