class_docstr: Main class from which all Flows should inherit.
  
  Attributes
  ----------
  index
  input
docstr: Helper function for merging artifacts in a join step.
  
  This function takes all the artifacts coming from the branches of a
  join point and assigns them to self in the calling step. Only artifacts
  not set in the current step are considered. If, for a given artifact, different
  values are present on the incoming edges, an error will be thrown and the artifacts
  that conflict will be reported.
  
  As a few examples, in the simple graph: A splitting into B and C and joining in D:
  ```
  A:
    self.x = 5
    self.y = 6
  B:
    self.b_var = 1
    self.x = from_b
  C:
    self.x = from_c
  
  D:
    merge_artifacts(inputs)
  ```
  In D, the following artifacts are set:
    - `y` (value: 6), `b_var` (value: 1)
    - if `from_b` and `from_c` are the same, `x` will be accessible and have value `from_b`
    - if `from_b` and `from_c` are different, an error will be thrown. To prevent this error,
      you need to manually set `self.x` in D to a merged value (for example the max) prior to
      calling `merge_artifacts`.
  
  Parameters
  ----------
  inputs : Inputs
      Incoming steps to the join point.
  exclude : List[str], optional
      If specified, do not consider merging artifacts with a name in `exclude`.
      Cannot specify if `include` is also specified.
  include : List[str], optional
      If specified, only merge artifacts specified. Cannot specify if `exclude` is
      also specified.
  
  Raises
  ------
  MetaflowException
      This exception is thrown if this is not called in a join step.
  UnhandledInMergeArtifactsException
      This exception is thrown in case of unresolved conflicts.
  MissingInMergeArtifactsException
      This exception is thrown in case an artifact specified in `include` cannot
      be found.
code: def merge_artifacts(
      self,
      inputs: Inputs,
      exclude: Optional[List[str]] = None,
      include: Optional[List[str]] = None,
  ) -> None:
      """
      Helper function for merging artifacts in a join step.
  
      This function takes all the artifacts coming from the branches of a
      join point and assigns them to self in the calling step. Only artifacts
      not set in the current step are considered. If, for a given artifact, different
      values are present on the incoming edges, an error will be thrown and the artifacts
      that conflict will be reported.
  
      As a few examples, in the simple graph: A splitting into B and C and joining in D:
      ```
      A:
        self.x = 5
        self.y = 6
      B:
        self.b_var = 1
        self.x = from_b
      C:
        self.x = from_c
  
      D:
        merge_artifacts(inputs)
      ```
      In D, the following artifacts are set:
        - `y` (value: 6), `b_var` (value: 1)
        - if `from_b` and `from_c` are the same, `x` will be accessible and have value `from_b`
        - if `from_b` and `from_c` are different, an error will be thrown. To prevent this error,
          you need to manually set `self.x` in D to a merged value (for example the max) prior to
          calling `merge_artifacts`.
  
      Parameters
      ----------
      inputs : Inputs
          Incoming steps to the join point.
      exclude : List[str], optional
          If specified, do not consider merging artifacts with a name in `exclude`.
          Cannot specify if `include` is also specified.
      include : List[str], optional
          If specified, only merge artifacts specified. Cannot specify if `exclude` is
          also specified.
  
      Raises
      ------
      MetaflowException
          This exception is thrown if this is not called in a join step.
      UnhandledInMergeArtifactsException
          This exception is thrown in case of unresolved conflicts.
      MissingInMergeArtifactsException
          This exception is thrown in case an artifact specified in `include` cannot
          be found.
      """
      include = include or []
      exclude = exclude or []
      node = self._graph[self._current_step]
      if node.type != "join":
          msg = (
              "merge_artifacts can only be called in a join and step *{step}* "
              "is not a join".format(step=self._current_step)
          )
          raise MetaflowException(msg)
      if len(exclude) > 0 and len(include) > 0:
          msg = "`exclude` and `include` are mutually exclusive in merge_artifacts"
          raise MetaflowException(msg)
  
      to_merge = {}
      unresolved = []
      for inp in inputs:
          # available_vars is the list of variables from inp that should be considered
          if include:
              available_vars = (
                  (var, sha)
                  for var, sha in inp._datastore.items()
                  if (var in include) and (not hasattr(self, var))
              )
          else:
              available_vars = (
                  (var, sha)
                  for var, sha in inp._datastore.items()
                  if (var not in exclude) and (not hasattr(self, var))
              )
          for var, sha in available_vars:
              _, previous_sha = to_merge.setdefault(var, (inp, sha))
              if previous_sha != sha:
                  # We have a conflict here
                  unresolved.append(var)
      # Check if everything in include is present in to_merge
      missing = []
      for v in include:
          if v not in to_merge and not hasattr(self, v):
              missing.append(v)
      if unresolved:
          # We have unresolved conflicts, so we do not set anything and error out
          msg = (
              "Step *{step}* cannot merge the following artifacts due to them "
              "having conflicting values:\n[{artifacts}].\nTo remedy this issue, "
              "be sure to explicitly set those artifacts (using "
              "self.<artifact_name> = ...) prior to calling merge_artifacts.".format(
                  step=self._current_step,
                  artifacts=", ".join(unresolved),
              )
          )
          raise UnhandledInMergeArtifactsException(msg, unresolved)
      if missing:
          msg = (
              "Step *{step}* specifies that [{include}] should be merged but "
              "[{missing}] are not present.\nTo remedy this issue, make sure "
              "that the values specified in only come from at least one branch".format(
                  step=self._current_step,
                  include=", ".join(include),
                  missing=", ".join(missing),
              )
          )
          raise MissingInMergeArtifactsException(msg, missing)
      # If things are resolved, we pass down the variables from the input datastores
      for var, (inp, _) in to_merge.items():
          self._datastore.passdown_partial(inp._datastore, [var])
  
imports: import inspect
  import os
  import sys
  import traceback
  from itertools import islice
  from types import FunctionType, MethodType
  from typing import Any, Callable, List, Optional, Tuple
  from .None import cmd_with_io
  from .parameters import DelayedEvaluationParameter, Parameter
  from .exception import MetaflowException, MissingInMergeArtifactsException, UnhandledInMergeArtifactsException
  from .graph import FlowGraph
  from .unbounded_foreach import UnboundedForeachInput
  from .datastore.inputs import Inputs
label_desc: Use image node_id 13 for calling the FlowSpec obj's underlying member method code with example usage: obj.merge_artifacts(inputs, exclude, include) without return types