(self, dag)
Expand DAG shorthand.

    Currently the only shorthand is "direct connection".
    A direct connection is a connection {a: b} where a.input_type is a dict,
    b.output_type is a dict, and a.input_type == b.output_type. This is not
    actually valid, but we can convert it to a valid connection.

    {a: b} is expanded to
    {a: {"name_1": b["name_1"], "name_2": b["name_2"], ...}}.
    {DagOutput(): {"name_1", obj1, "name_2": obj2, ...} is expanded to
    {DagOutput("name_1"): obj1, DagOutput("name_2"): obj2, ...}.

    Args:
      dag: A dictionary encoding the DAG.

    Yields:
      Key, value pairs for a new dag dictionary.

    Raises:
      InvalidDictionaryOutputError: If `DagOutput` is not used correctly.
Expand DAG shorthand.
Expand DAG shorthand .
def _expand_dag_shorthands(self, dag):
    """Expand DAG shorthand.

    Currently the only shorthand is "direct connection".
    A direct connection is a connection {a: b} where a.input_type is a dict,
    b.output_type is a dict, and a.input_type == b.output_type. This is not
    actually valid, but we can convert it to a valid connection.

    {a: b} is expanded to
    {a: {"name_1": b["name_1"], "name_2": b["name_2"], ...}}.
    {DagOutput(): {"name_1", obj1, "name_2": obj2, ...} is expanded to
    {DagOutput("name_1"): obj1, DagOutput("name_2"): obj2, ...}.

    Args:
      dag: A dictionary encoding the DAG.

    Yields:
      Key, value pairs for a new dag dictionary.

    Raises:
      InvalidDictionaryOutputError: If `DagOutput` is not used correctly.
    """
    for key, val in dag.items():
      # Direct connection.
      if (isinstance(key, pipeline.Pipeline) and
          isinstance(val, pipeline.Pipeline) and
          isinstance(key.input_type, dict) and
          key.input_type == val.output_type):
        yield key, dict((name, val[name]) for name in val.output_type)
      elif key == DagOutput():
        if (isinstance(val, pipeline.Pipeline) and
            isinstance(val.output_type, dict)):
          dependency = [(name, val[name]) for name in val.output_type]
        elif isinstance(val, dict):
          dependency = val.items()
        else:
          raise InvalidDictionaryOutputError(
              'DagOutput() with no name can only be connected to a dictionary '
              'or a Pipeline whose output_type is a dictionary. Found '
              'DagOutput() connected to %s' % val)
        for name, subordinate in dependency:
          yield DagOutput(name), subordinate
      elif isinstance(key, DagOutput):
        if isinstance(val, dict):
          raise InvalidDictionaryOutputError(
              'DagOutput("%s") which has name "%s" can only be connected to a '
              'single input, not dictionary %s. Use DagOutput() without name '
              'instead.' % (key.name, key.name, val))
        if (isinstance(val, pipeline.Pipeline) and
            isinstance(val.output_type, dict)):
          raise InvalidDictionaryOutputError(
              'DagOutput("%s") which has name "%s" can only be connected to a '
              'single input, not pipeline %s which has dictionary '
              'output_type %s. Use DagOutput() without name instead.'
              % (key.name, key.name, val, val.output_type))
        yield key, val
      else:
        yield key, val
def _expand_dag_shorthands ( self , dag ) : for key , val in dag . items ( ) : # Direct connection. if ( isinstance ( key , pipeline . Pipeline ) and isinstance ( val , pipeline . Pipeline ) and isinstance ( key . input_type , dict ) and key . input_type == val . output_type ) : yield key , dict ( ( name , val [ name ] ) for name in val . output_type ) elif key == DagOutput ( ) : if ( isinstance ( val , pipeline . Pipeline ) and isinstance ( val . output_type , dict ) ) : dependency = [ ( name , val [ name ] ) for name in val . output_type ] elif isinstance ( val , dict ) : dependency = val . items ( ) else : raise InvalidDictionaryOutputError ( 'DagOutput() with no name can only be connected to a dictionary ' 'or a Pipeline whose output_type is a dictionary. Found ' 'DagOutput() connected to %s' % val ) for name , subordinate in dependency : yield DagOutput ( name ) , subordinate elif isinstance ( key , DagOutput ) : if isinstance ( val , dict ) : raise InvalidDictionaryOutputError ( 'DagOutput("%s") which has name "%s" can only be connected to a ' 'single input, not dictionary %s. Use DagOutput() without name ' 'instead.' % ( key . name , key . name , val ) ) if ( isinstance ( val , pipeline . Pipeline ) and isinstance ( val . output_type , dict ) ) : raise InvalidDictionaryOutputError ( 'DagOutput("%s") which has name "%s" can only be connected to a ' 'single input, not pipeline %s which has dictionary ' 'output_type %s. Use DagOutput() without name instead.' % ( key . name , key . name , val , val . output_type ) ) yield key , val else : yield key , val