Method edge

graphlearn/python/graph.py:276–318

Add graph edges that will be loaded from a given path. Args: source (string): Data source path where to load the edges. edge_type (tuple): A tuple of (src_type, dst_type, edge_type) that indicates types of the edges. decoder (Decoder): A Decoder object to describe the

(self,
           source,
           edge_type,
           decoder=None,
           directed=True,
           option=None,
           mask=utils.Mask.NONE)

    return self

  def edge(self,
           source,
           edge_type,
           decoder=None,
           directed=True,
           option=None,
           mask=utils.Mask.NONE):
    """ Add graph edges that will be loaded from a given path.

    Args:
      source (string): Data source path where to load the edges.
      edge_type (tuple): A tuple of (src_type, dst_type, edge_type) that
        indicates types of the edges.
      decoder (Decoder): A Decoder object to describe the data schema.
      directed (boolean): Whether edges are directed.
      mask (TRAIN | TEST | VAL): Mark the source as TRAIN data, TEST data
        or VAL data for the given edge_type in the graph.
    """
    if not isinstance(source, str):
      raise ValueError('source for edge() must be a string.')
    if not isinstance(edge_type, tuple) or len(edge_type) != 3:
      raise ValueError("edge_type for edge() must be a tuple of "
                       "(src_type, dst_tye, edge_type).")
    if not decoder:
      decoder = data.Decoder()
    if not isinstance(decoder, data.Decoder):
      raise ValueError('decoder must be an instance of Decoder, got {}'
                       .format(type(decoder)))

    masked_edge_type = utils.get_mask_type(edge_type[2], mask)
    self._edge_decoders[masked_edge_type] = decoder

    self._topology.add(masked_edge_type, edge_type[0], edge_type[1])
    edge_source = self._construct_edge_source(
        source, (edge_type[0], edge_type[1], masked_edge_type),
        decoder,
        direction=pywrap.Direction.ORIGIN,
        option=option)
    self._edge_sources.append(edge_source)

    if not directed:
      self.add_reverse_edges(edge_type, source, decoder, option)
    return self

  def _copy_edge_source(self, edge):
    result = pywrap.EdgeSource()