def check_queue_exists()

in common/cloud_task/cloud_task_publisher.py [0:0]

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

    def check_queue_exists(self) -> bool:
        """
        Checks if a queue exists.
        """
        try:
            self._cloud_task_client.get_queue(name=self._queue_fqn)
            return True
        except NotFound:
            self._logger.info(f"Queue {self._queue_fqn} does not exist. "
                              f"Queue will be created automatically.")
            return False