airflow/contrib/hooks/gcp_pubsub_hook.py
def delete_topic(self, project, topic, fail_if_not_exists=False):
        """Deletes a Pub/Sub topic if it exists.

        :param project: the GCP project ID in which to delete the topic
        :type project: str
        :param topic: the Pub/Sub topic name to delete; do not
            include the ``projects/{project}/topics/`` prefix.
        :type topic: str
        :param fail_if_not_exists: if set, raise an exception if the topic
            does not exist
        :type fail_if_not_exists: bool
        """
        service = self.get_conn()
        full_topic = _format_topic(project, topic)
        try:
            service.projects().topics().delete(topic=full_topic).execute(num_retries=self.num_retries)
        except HttpError as e:
            # Status code 409 indicates that the topic was not found
            if str(e.resp['status']) == '404':
                message = 'Topic does not exist: {}'.format(full_topic)
                self.log.warning(message)
                if fail_if_not_exists:
                    raise PubSubException(message)
            else:
                raise PubSubException(
                    'Error deleting topic {}'.format(full_topic), e)
def delete_topic(self, project, topic, fail_if_not_exists=False):
        """Deletes a Pub/Sub topic if it exists.

        :param project: the GCP project ID in which to delete the topic
        :type project: str
        :param topic: the Pub/Sub topic name to delete; do not
            include the ``projects/{project}/topics/`` prefix.
        :type topic: str
        :param fail_if_not_exists: if set, raise an exception if the topic
            does not exist
        :type fail_if_not_exists: bool
        """
        service = self.get_conn()
        full_topic = _format_topic(project, topic)
        try:
            service.projects().topics().delete(topic=full_topic).execute(num_retries=self.num_retries)
        except HttpError as e:
            # Status code 409 indicates that the topic was not found
            if str(e.resp['status']) == '404':
                message = 'Topic does not exist: {}'.format(full_topic)
                self.log.warning(message)
                if fail_if_not_exists:
                    raise PubSubException(message)
            else:
                raise PubSubException(
                    'Error deleting topic {}'.format(full_topic), e)
Deletes a Pub/Sub topic if it exists.

        :param project: the GCP project ID in which to delete the topic
        :type project: str
        :param topic: the Pub/Sub topic name to delete; do not
            include the ``projects/{project}/topics/`` prefix.
        :type topic: str
        :param fail_if_not_exists: if set, raise an exception if the topic
            does not exist
        :type fail_if_not_exists: bool