def _add_node_attribute()

in tfx_airflow/notebooks/utils.py [0:0]

• 13 lines of code
• 4 McCabe index (conditional complexity)

  def _add_node_attribute(self, g, node_id, depth, is_artifact):
    """Adds the attributes of given artifact or execution to the graph `g`."""
    # if it is not an artifact, use negative gnode id
    gnode_id = node_id if is_artifact else -1 * node_id
    g.add_node(gnode_id, depth=depth, is_artifact=is_artifact)
    node_label = ''
    if is_artifact:
      [a] = self.metadata_store.get_artifacts_by_id([node_id])
      [t] = self.metadata_store.get_artifact_types_by_id([a.type_id])
      node_label += t.name
    else:
      [e] = self.metadata_store.get_executions_by_id([node_id])
      [t] = self.metadata_store.get_execution_types_by_id([e.type_id])
      node_label += t.name
    g.nodes[gnode_id]['_label_'] = node_label