def _verify_outputs()

in cli/src/klio_cli/commands/job/verify.py [0:0]

• 45 lines of code
• 11 McCabe index (conditional complexity)

    def _verify_outputs(self):
        unverified_bucket_count = 0
        unverified_topic_count = 0

        data_outputs = self.klio_config.job_config.data.outputs

        logging.info("Verifying your data outputs...")

        for output in data_outputs:
            output_bucket = output.location

            if output_bucket is not None:
                verified_bucket = self._verify_gcs_bucket(output_bucket)
                if not verified_bucket:
                    unverified_bucket_count += 1
            else:
                logging.error(
                    "There is no data_location for {}".format(output)
                )

        event_outputs = self.klio_config.job_config.events.outputs

        logging.info("Verifying your event outputs...")

        pubsub_event_outputs = [
            _i for _i in event_outputs if _i.name == "pubsub"
        ]
        for output in pubsub_event_outputs:
            output_topic = output.topic

            if output_topic is not None:
                verified_topic = self._verify_pub_topic(output_topic, "output")
                if not verified_topic:
                    unverified_topic_count += 1
            else:
                logging.error("There is no topic for {}".format(output))

        if pubsub_event_outputs:
            logging.info(
                "You have {} unverified output buckets "
                "and {} unverified output "
                "topics".format(
                    unverified_bucket_count, unverified_topic_count
                )
            )
        else:
            logging.info(
                "You have {} unverified output buckets".format(
                    unverified_bucket_count
                )
            )
        if unverified_bucket_count + unverified_topic_count == 0:
            return True
        return False