def _check_existing_label()

in transcoder/output/google_cloud/PubSubOutputManager.py [0:0]

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

    def _check_existing_label(self, _existing_topic):
        if GOOGLE_PACKAGED_SOLUTION_KEY not in _existing_topic.labels \
                or _existing_topic.labels.get(GOOGLE_PACKAGED_SOLUTION_KEY, None) != GOOGLE_PACKAGED_SOLUTION_VALUE:
            topic = Topic()
            topic.name = _existing_topic.name
            topic.labels[  # pylint: disable=unsupported-assignment-operation
                GOOGLE_PACKAGED_SOLUTION_KEY] = GOOGLE_PACKAGED_SOLUTION_VALUE
            request = UpdateTopicRequest(
                topic=topic,
                update_mask={"paths": ["labels"]}
            )
            try:
                self.publisher.update_topic(request=request)
            except Exception as err:  # pylint: disable=broad-except
                logging.warning("Failed to update topic labels: %s", err)