def _generate_report()

in ilm/ilm_lib/dicom_store_lib.py [0:0]

• 86 lines of code
• 8 McCabe index (conditional complexity)

  def _generate_report(self):
    """Generates report of storage class updates.

    Reports instances' updates and status (success, failure, still running) for
    each move rule and condition in ILM config.
    """
    self._compute_results()
    csv_separator = ','
    header = csv_separator.join([
        'from_storage_class',
        'to_storage_class',
        'condition',
        'rule_index',
        'condition_index',
        'succeeded',
        'failed',
        'unfinished',
    ])
    summarized_report_results = [header]
    detailed_report_results = [header]
    logging.info('Generating storage class updates report(s).')
    # TODO: Also add operation name.
    for rule_index, rule in enumerate(self._move_rules):
      from_storage_class = rule.from_storage_class
      to_storage_class = rule.to_storage_class
      for condition_index, condition in enumerate(rule.conditions):
        rule_id = ilm_types.MoveRuleId(
            rule_index=rule_index, condition_index=condition_index
        )
        if rule_id not in self._storage_class_results:
          continue
        condition_results = self._storage_class_results[rule_id]
        logging.info(
            'Instances moved from %s to %s due to %s (move rule %s, condition '
            '%s): %s succeeded, %s failed, %s unfinished',
            from_storage_class.value,
            to_storage_class.value,
            condition,
            rule_index,
            condition_index,
            condition_results.succeeded_count,
            condition_results.failed_count,
            condition_results.unfinished_count,
        )
        summarized_report_results.append(
            csv_separator.join([
                str(i)
                for i in [
                    from_storage_class.value,
                    to_storage_class.value,
                    condition,
                    rule_index,
                    condition_index,
                    condition_results.succeeded_count,
                    condition_results.failed_count,
                    condition_results.unfinished_count,
                ]
            ]),
        )
        if self._detailed_report:
          detailed_report_results.append(
              csv_separator.join([
                  str(i)
                  for i in [
                      from_storage_class.value,
                      to_storage_class.value,
                      condition,
                      rule_index,
                      condition_index,
                      f'"{condition_results.succeeded_instances}"',
                      f'"{condition_results.failed_instances}"',
                      f'"{condition_results.unfinished_instances}"',
                  ]
              ])
          )
    summarized_report_results_str = '\n'.join(summarized_report_results)
    pipeline_util.write_gcs_file(
        file_content=summarized_report_results_str,
        gcs_uri=self._summarized_report,
    )
    logging.info(
        'Wrote summarized report of updates to %s.', self._summarized_report
    )
    if self._detailed_report:
      detailed_report_results_str = '\n'.join(detailed_report_results)
      pipeline_util.write_gcs_file(
          file_content=detailed_report_results_str,
          gcs_uri=self._detailed_report,
      )
      logging.info(
          'Wrote detailed report of updates to %s.', self._detailed_report
      )