def get_status_tree(root_pipeline_id):
  """Gets the full status tree of a pipeline.

  Args:
    root_pipeline_id: The pipeline ID to get status for.

  Returns:
    Dictionary with the keys:
      rootPipelineId: The ID of the root pipeline.
      slots: Mapping of slot IDs to result of from _get_internal_slot.
      pipelines: Mapping of pipeline IDs to result of _get_internal_status.

  Raises:
    PipelineStatusError if any input is bad.
  """
  root_pipeline_key = db.Key.from_path(_PipelineRecord.kind(), root_pipeline_id)
  root_pipeline_record = db.get(root_pipeline_key)
  if root_pipeline_record is None:
    raise PipelineStatusError(
        'Could not find pipeline ID "%s"' % root_pipeline_id)

  # If the supplied root_pipeline_id is not actually the root pipeline that's
  # okay. We'll find the real root and override the value they passed in.
  actual_root_key = _PipelineRecord.root_pipeline.get_value_for_datastore(
      root_pipeline_record)
  if actual_root_key != root_pipeline_key:
    root_pipeline_key = actual_root_key
    root_pipeline_id = root_pipeline_key.id_or_name()
    root_pipeline_record = db.get(root_pipeline_key)
    if not root_pipeline_record:
      raise PipelineStatusError(
          'Could not find pipeline ID "%s"' % root_pipeline_id)

  # Run all queries asynchronously.
  queries = {}
  for model in (_PipelineRecord, _SlotRecord, _BarrierRecord, _StatusRecord):
    queries[model] = model.all().filter(
        'root_pipeline =', root_pipeline_key).run(batch_size=1000)

  found_pipeline_dict = dict(
      (stage.key(), stage) for stage in queries[_PipelineRecord])
  found_slot_dict = dict(
      (slot.key(), slot) for slot in queries[_SlotRecord])
  found_barrier_dict = dict(
      (barrier.key(), barrier) for barrier in queries[_BarrierRecord])
  found_status_dict = dict(
      (status.key(), status) for status in queries[_StatusRecord])

  # Breadth-first traversal of _PipelineRecord instances by following
  # _PipelineRecord.fanned_out property values.
  valid_pipeline_keys = set([root_pipeline_key])
  slot_filler_dict = {}  # slot_key to pipeline_key
  expand_stack = [root_pipeline_record]
  while expand_stack:
    old_stack = expand_stack
    expand_stack = []
    for pipeline_record in old_stack:
      for child_pipeline_key in pipeline_record.fanned_out:
        # This will let us prune off those pipelines which were allocated in
        # the Datastore but were never run due to mid-flight task failures.
        child_pipeline_record = found_pipeline_dict.get(child_pipeline_key)
        if child_pipeline_record is None:
          raise PipelineStatusError(
              'Pipeline ID "%s" points to child ID "%s" which does not exist.'
              % (pipeline_record.key().name(), child_pipeline_key.name()))
        expand_stack.append(child_pipeline_record)
        valid_pipeline_keys.add(child_pipeline_key)

        # Figure out the deepest pipeline that's responsible for outputting to
        # a particular _SlotRecord, so we can report which pipeline *should*
        # be the filler.
        child_outputs = child_pipeline_record.params['output_slots']
        for output_slot_key in child_outputs.itervalues():
          slot_filler_dict[db.Key(output_slot_key)] = child_pipeline_key

  output = {
    'rootPipelineId': root_pipeline_id,
    'slots': {},
    'pipelines': {},
  }

  for pipeline_key in found_pipeline_dict.keys():
    if pipeline_key not in valid_pipeline_keys:
      continue
    output['pipelines'][pipeline_key.name()] = _get_internal_status(
        pipeline_key=pipeline_key,
        pipeline_dict=found_pipeline_dict,
        slot_dict=found_slot_dict,
        barrier_dict=found_barrier_dict,
        status_dict=found_status_dict)

  for slot_key, filler_pipeline_key in slot_filler_dict.iteritems():
    output['slots'][str(slot_key)] = _get_internal_slot(
        slot_key=slot_key,
        filler_pipeline_key=filler_pipeline_key,
        slot_dict=found_slot_dict)

  return output
def get_status_tree ( root_pipeline_id ) : root_pipeline_key = db . Key . from_path ( _PipelineRecord . kind ( ) , root_pipeline_id ) root_pipeline_record = db . get ( root_pipeline_key ) if root_pipeline_record is None : raise PipelineStatusError ( 'Could not find pipeline ID "%s"' % root_pipeline_id ) actual_root_key = _PipelineRecord . root_pipeline . get_value_for_datastore ( root_pipeline_record ) if actual_root_key != root_pipeline_key : root_pipeline_key = actual_root_key root_pipeline_id = root_pipeline_key . id_or_name ( ) root_pipeline_record = db . get ( root_pipeline_key ) if not root_pipeline_record : raise PipelineStatusError ( 'Could not find pipeline ID "%s"' % root_pipeline_id ) queries = { } for model in ( _PipelineRecord , _SlotRecord , _BarrierRecord , _StatusRecord ) : queries [ model ] = model . all ( ) . filter ( 'root_pipeline =' , root_pipeline_key ) . run ( batch_size = 1000 ) found_pipeline_dict = dict ( ( stage . key ( ) , stage ) for stage in queries [ _PipelineRecord ] ) found_slot_dict = dict ( ( slot . key ( ) , slot ) for slot in queries [ _SlotRecord ] ) found_barrier_dict = dict ( ( barrier . key ( ) , barrier ) for barrier in queries [ _BarrierRecord ] ) found_status_dict = dict ( ( status . key ( ) , status ) for status in queries [ _StatusRecord ] ) valid_pipeline_keys = set ( [ root_pipeline_key ] ) slot_filler_dict = { } expand_stack = [ root_pipeline_record ] while expand_stack : old_stack = expand_stack expand_stack = [ ] for pipeline_record in old_stack : for child_pipeline_key in pipeline_record . fanned_out : child_pipeline_record = found_pipeline_dict . get ( child_pipeline_key ) if child_pipeline_record is None : raise PipelineStatusError ( 'Pipeline ID "%s" points to child ID "%s" which does not exist.' % ( pipeline_record . key ( ) . name ( ) , child_pipeline_key . name ( ) ) ) expand_stack . append ( child_pipeline_record ) valid_pipeline_keys . add ( child_pipeline_key ) child_outputs = child_pipeline_record . params [ 'output_slots' ] for output_slot_key in child_outputs . itervalues ( ) : slot_filler_dict [ db . Key ( output_slot_key ) ] = child_pipeline_key output = { 'rootPipelineId' : root_pipeline_id , 'slots' : { } , 'pipelines' : { } , } for pipeline_key in found_pipeline_dict . keys ( ) : if pipeline_key not in valid_pipeline_keys : continue output [ 'pipelines' ] [ pipeline_key . name ( ) ] = _get_internal_status ( pipeline_key = pipeline_key , pipeline_dict = found_pipeline_dict , slot_dict = found_slot_dict , barrier_dict = found_barrier_dict , status_dict = found_status_dict ) for slot_key , filler_pipeline_key in slot_filler_dict . iteritems ( ) : output [ 'slots' ] [ str ( slot_key ) ] = _get_internal_slot ( slot_key = slot_key , filler_pipeline_key = filler_pipeline_key , slot_dict = found_slot_dict ) return output
Gets the full status tree of a pipeline.

  Args:
    root_pipeline_id: The pipeline ID to get status for.

  Returns:
    Dictionary with the keys:
      rootPipelineId: The ID of the root pipeline.
      slots: Mapping of slot IDs to result of from _get_internal_slot.
      pipelines: Mapping of pipeline IDs to result of _get_internal_status.

  Raises:
    PipelineStatusError if any input is bad.