airflow/contrib/hooks/gcp_pubsub_hook.py
def create_subscription(self, topic_project, topic, subscription=None,
                            subscription_project=None, ack_deadline_secs=10,
                            fail_if_exists=False):
        """Creates a Pub/Sub subscription, if it does not already exist.

        :param topic_project: the GCP project ID of the topic that the
            subscription will be bound to.
        :type topic_project: str
        :param topic: the Pub/Sub topic name that the subscription will be bound
            to create; do not include the ``projects/{project}/subscriptions/``
            prefix.
        :type topic: str
        :param subscription: the Pub/Sub subscription name. If empty, a random
            name will be generated using the uuid module
        :type subscription: str
        :param subscription_project: the GCP project ID where the subscription
            will be created. If unspecified, ``topic_project`` will be used.
        :type subscription_project: str
        :param ack_deadline_secs: Number of seconds that a subscriber has to
            acknowledge each message pulled from the subscription
        :type ack_deadline_secs: int
        :param fail_if_exists: if set, raise an exception if the topic
            already exists
        :type fail_if_exists: bool
        :return: subscription name which will be the system-generated value if
            the ``subscription`` parameter is not supplied
        :rtype: str
        """
        service = self.get_conn()
        full_topic = _format_topic(topic_project, topic)
        if not subscription:
            subscription = 'sub-{}'.format(uuid4())
        if not subscription_project:
            subscription_project = topic_project
        full_subscription = _format_subscription(subscription_project,
                                                 subscription)
        body = {
            'topic': full_topic,
            'ackDeadlineSeconds': ack_deadline_secs
        }
        try:
            service.projects().subscriptions().create(
                name=full_subscription, body=body).execute(num_retries=self.num_retries)
        except HttpError as e:
            # Status code 409 indicates that the subscription already exists.
            if str(e.resp['status']) == '409':
                message = 'Subscription already exists: {}'.format(
                    full_subscription)
                self.log.warning(message)
                if fail_if_exists:
                    raise PubSubException(message)
            else:
                raise PubSubException(
                    'Error creating subscription {}'.format(full_subscription),
                    e)
        return subscription
def create_subscription(self, topic_project, topic, subscription=None,
                            subscription_project=None, ack_deadline_secs=10,
                            fail_if_exists=False):
        """Creates a Pub/Sub subscription, if it does not already exist.

        :param topic_project: the GCP project ID of the topic that the
            subscription will be bound to.
        :type topic_project: str
        :param topic: the Pub/Sub topic name that the subscription will be bound
            to create; do not include the ``projects/{project}/subscriptions/``
            prefix.
        :type topic: str
        :param subscription: the Pub/Sub subscription name. If empty, a random
            name will be generated using the uuid module
        :type subscription: str
        :param subscription_project: the GCP project ID where the subscription
            will be created. If unspecified, ``topic_project`` will be used.
        :type subscription_project: str
        :param ack_deadline_secs: Number of seconds that a subscriber has to
            acknowledge each message pulled from the subscription
        :type ack_deadline_secs: int
        :param fail_if_exists: if set, raise an exception if the topic
            already exists
        :type fail_if_exists: bool
        :return: subscription name which will be the system-generated value if
            the ``subscription`` parameter is not supplied
        :rtype: str
        """
        service = self.get_conn()
        full_topic = _format_topic(topic_project, topic)
        if not subscription:
            subscription = 'sub-{}'.format(uuid4())
        if not subscription_project:
            subscription_project = topic_project
        full_subscription = _format_subscription(subscription_project,
                                                 subscription)
        body = {
            'topic': full_topic,
            'ackDeadlineSeconds': ack_deadline_secs
        }
        try:
            service.projects().subscriptions().create(
                name=full_subscription, body=body).execute(num_retries=self.num_retries)
        except HttpError as e:
            # Status code 409 indicates that the subscription already exists.
            if str(e.resp['status']) == '409':
                message = 'Subscription already exists: {}'.format(
                    full_subscription)
                self.log.warning(message)
                if fail_if_exists:
                    raise PubSubException(message)
            else:
                raise PubSubException(
                    'Error creating subscription {}'.format(full_subscription),
                    e)
        return subscription
def create_subscription ( self , topic_project , topic , subscription = None , subscription_project = None , ack_deadline_secs = 10 , fail_if_exists = False ) : service = self . get_conn ( ) full_topic = _format_topic ( topic_project , topic ) if not subscription : subscription = 'sub-{}' . format ( uuid4 ( ) ) if not subscription_project : subscription_project = topic_project full_subscription = _format_subscription ( subscription_project , subscription ) body = { 'topic' : full_topic , 'ackDeadlineSeconds' : ack_deadline_secs } try : service . projects ( ) . subscriptions ( ) . create ( name = full_subscription , body = body ) . execute ( num_retries = self . num_retries ) except HttpError as e : # Status code 409 indicates that the subscription already exists. if str ( e . resp [ 'status' ] ) == '409' : message = 'Subscription already exists: {}' . format ( full_subscription ) self . log . warning ( message ) if fail_if_exists : raise PubSubException ( message ) else : raise PubSubException ( 'Error creating subscription {}' . format ( full_subscription ) , e ) return subscription
Creates a Pub/Sub subscription, if it does not already exist.

        :param topic_project: the GCP project ID of the topic that the
            subscription will be bound to.
        :type topic_project: str
        :param topic: the Pub/Sub topic name that the subscription will be bound
            to create; do not include the ``projects/{project}/subscriptions/``
            prefix.
        :type topic: str
        :param subscription: the Pub/Sub subscription name. If empty, a random
            name will be generated using the uuid module
        :type subscription: str
        :param subscription_project: the GCP project ID where the subscription
            will be created. If unspecified, ``topic_project`` will be used.
        :type subscription_project: str
        :param ack_deadline_secs: Number of seconds that a subscriber has to
            acknowledge each message pulled from the subscription
        :type ack_deadline_secs: int
        :param fail_if_exists: if set, raise an exception if the topic
            already exists
        :type fail_if_exists: bool
        :return: subscription name which will be the system-generated value if
            the ``subscription`` parameter is not supplied
        :rtype: str
Creates a Pub / Sub subscription if it does not already exist .