def _get_resources()

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

• 32 lines of code
• 12 McCabe index (conditional complexity)

    def _get_resources(self):
        # Go through the list of resources set up
        # in input/output and ask for user input
        # on each one that we should delete.
        # The following are the types of resources we are considering:
        # 1. topics
        # 2. subscriptions
        # 3. buckets
        to_delete = {
            resource_type: [] for resource_type in self.resource_types
        }
        to_delete["stackdriver_group"] = False

        ev_inputs = self._job_config.events.inputs
        ev_outputs = self._job_config.events.outputs
        for resource in ev_inputs + ev_outputs:
            if "pubsub" == resource.name:
                if self._confirmation_dialog("topic", resource.topic):
                    to_delete["topic"].append(resource.topic)

                if not hasattr(resource, "subscription"):
                    continue
                if self._confirmation_dialog(
                    "subscription", resource.subscription
                ):
                    to_delete["subscription"].append(resource.subscription)

        data_inputs = self._job_config.data.inputs
        data_outputs = self._job_config.data.outputs
        for resource in data_inputs + data_outputs:
            if "gcs" == resource.name:
                if self._confirmation_dialog("location", resource.location):
                    to_delete["location"].append(resource.location)

        # Now lets handle the stackdriver group
        _, dashboard_name = sd_utils.generate_group_meta(
            self._pipeline_options.project,
            self.config.job_name,
            self._pipeline_options.region,
        )
        to_delete["stackdriver_group"] = self._confirmation_dialog(
            "stackdriver dashboard group", dashboard_name
        )

        return to_delete