class_docstr: Hook for accessing Google Pub/Sub.
  
  The Google Cloud project against which actions are applied is determined by
  the project embedded in the Connection referenced by gcp_conn_id.
class_docstr_tok: ["Hook","for","accessing","Google","Pub\/Sub",".","The","Google","Cloud","project","against","which","actions","are","applied","is","determined","by","the","project","embedded","in","the","Connection","referenced","by","gcp_conn_id","."]
docstr: Creates a Pub/Sub subscription, if it does not already exist.
  
  :param topic: the Pub/Sub topic name that the subscription will be bound
      to create; do not include the ``projects/{project}/subscriptions/`` prefix.
  :param project_id: Optional, the Google Cloud project ID of the topic that the subscription will be
      bound to. If set to None or missing, the default project_id from the Google Cloud connection
      is used.
  :param subscription: the Pub/Sub subscription name. If empty, a random
      name will be generated using the uuid module
  :param subscription_project_id: the Google Cloud project ID where the subscription
      will be created. If unspecified, ``project_id`` will be used.
  :param ack_deadline_secs: Number of seconds that a subscriber has to
      acknowledge each message pulled from the subscription
  :param fail_if_exists: if set, raise an exception if the topic
      already exists
  :param push_config: If push delivery is used with this subscription,
      this field is used to configure it. An empty ``pushConfig`` signifies
      that the subscriber will pull and ack messages using API methods.
  :param retain_acked_messages: Indicates whether to retain acknowledged
      messages. If true, then messages are not expunged from the subscription's
      backlog, even if they are acknowledged, until they fall out of the
      ``message_retention_duration`` window. This must be true if you would
      like to Seek to a timestamp.
  :param message_retention_duration: How long to retain unacknowledged messages
      in the subscription's backlog, from the moment a message is published. If
      ``retain_acked_messages`` is true, then this also configures the
      retention of acknowledged messages, and thus configures how far back in
      time a ``Seek`` can be done. Defaults to 7 days. Cannot be more than 7
      days or less than 10 minutes.
  :param labels: Client-assigned labels; see
      https://cloud.google.com/pubsub/docs/labels
  :param enable_message_ordering: If true, messages published with the same
      ordering_key in PubsubMessage will be delivered to the subscribers in the order
      in which they are received by the Pub/Sub system. Otherwise, they may be
      delivered in any order.
  :param expiration_policy: A policy that specifies the conditions for this
      subscription's expiration. A subscription is considered active as long as any
      connected subscriber is successfully consuming messages from the subscription or
      is issuing operations on the subscription. If expiration_policy is not set,
      a default policy with ttl of 31 days will be used. The minimum allowed value for
      expiration_policy.ttl is 1 day.
  :param filter_: An expression written in the Cloud Pub/Sub filter language. If
      non-empty, then only PubsubMessages whose attributes field matches the filter are
      delivered on this subscription. If empty, then no messages are filtered out.
  :param dead_letter_policy: A policy that specifies the conditions for dead lettering
      messages in this subscription. If dead_letter_policy is not set, dead lettering is
      disabled.
  :param retry_policy: A policy that specifies how Pub/Sub retries message delivery
      for this subscription. If not set, the default retry policy is applied. This
      generally implies that messages will be retried as soon as possible for healthy
      subscribers. RetryPolicy will be triggered on NACKs or acknowledgement deadline
      exceeded events for a given message.
  :param retry: (Optional) A retry object used to retry requests.
      If None is specified, requests will not be retried.
  :param timeout: (Optional) The amount of time, in seconds, to wait for the request
      to complete. Note that if retry is specified, the timeout applies to each
      individual attempt.
  :param metadata: (Optional) Additional metadata that is provided to the method.
  :return: subscription name which will be the system-generated value if
      the ``subscription`` parameter is not supplied
docstr_tok: ["Creates","a","Pub\/Sub","subscription",",","if","it","does","not","already","exist",".",":","param","topic",":","the","Pub\/Sub","topic","name","that","the","subscription","will","be","bound","to","create",";","do","not","include","the","``","projects\/","{","project","}","\/subscriptions\/","``","prefix",".",":","param","project_id",":","Optional",",","the","Google","Cloud","project","ID","of","the","topic","that","the","subscription","will","be","bound","to",".","If","set","to","None","or","missing",",","the","default","project_id","from","the","Google","Cloud","connection","is","used",".",":","param","subscription",":","the","Pub\/Sub","subscription","name",".","If","empty",",","a","random","name","will","be","generated","using","the","uuid","module",":","param","subscription_project_id",":","the","Google","Cloud","project","ID","where","the","subscription","will","be","created",".","If","unspecified",",","``","project_id","``","will","be","used",".",":","param","ack_deadline_secs",":","Number","of","seconds","that","a","subscriber","has","to","acknowledge","each","message","pulled","from","the","subscription",":","param","fail_if_exists",":","if","set",",","raise","an","exception","if","the","topic","already","exists",":","param","push_config",":","If","push","delivery","is","used","with","this","subscription",",","this","field","is","used","to","configure","it",".","An","empty","``","pushConfig","``","signifies","that","the","subscriber","will","pull","and","ack","messages","using","API","methods",".",":","param","retain_acked_messages",":","Indicates","whether","to","retain","acknowledged","messages",".","If","true",",","then","messages","are","not","expunged","from","the","subscription's","backlog",",","even","if","they","are","acknowledged",",","until","they","fall","out","of","the","``","message_retention_duration","``","window",".","This","must","be","true","if","you","would","like","to","Seek","to","a","timestamp",".",":","param","message_retention_duration",":","How","long","to","retain","unacknowledged","messages","in","the","subscription","'s","backlog",",","from","the","moment","a","message","is","published",".","If","``","retain_acked_messages","``","is","true",",","then","this","also","configures","the","retention","of","acknowledged","messages",",","and","thus","configures","how","far","back","in","time","a","``","Seek","``","can","be","done",".","Defaults","to","7","days",".","Can","not","be","more","than","7","days","or","less","than","10","minutes",".",":","param","labels",":","Client-assigned","labels",";","see","https",":","\/\/cloud.google.com\/pubsub\/docs\/labels",":","param","enable_message_ordering",":","If","true",",","messages","published","with","the","same","ordering_key","in","PubsubMessage","will","be","delivered","to","the","subscribers","in","the","order","in","which","they","are","received","by","the","Pub\/Sub","system",".","Otherwise",",","they","may","be","delivered","in","any","order",".",":","param","expiration_policy",":","A","policy","that","specifies","the","conditions","for","this","subscription","'s","expiration",".","A","subscription","is","considered","active","as","long","as","any","connected","subscriber","is","successfully","consuming","messages","from","the","subscription","or","is","issuing","operations","on","the","subscription",".","If","expiration_policy","is","not","set",",","a","default","policy","with","ttl","of","31","days","will","be","used",".","The","minimum","allowed","value","for","expiration_policy.ttl","is","1","day",".",":","param","filter_",":","An","expression","written","in","the","Cloud","Pub\/Sub","filter","language",".","If","non-empty",",","then","only","PubsubMessages","whose","attributes","field","matches","the","filter","are","delivered","on","this","subscription",".","If","empty",",","then","no","messages","are","filtered","out",".",":","param","dead_letter_policy",":","A","policy","that","specifies","the","conditions","for","dead","lettering","messages","in","this","subscription",".","If","dead_letter_policy","is","not","set",",","dead","lettering","is","disabled",".",":","param","retry_policy",":","A","policy","that","specifies","how","Pub\/Sub","retries","message","delivery","for","this","subscription",".","If","not","set",",","the","default","retry","policy","is","applied",".","This","generally","implies","that","messages","will","be","retried","as","soon","as","possible","for","healthy","subscribers",".","RetryPolicy","will","be","triggered","on","NACKs","or","acknowledgement","deadline","exceeded","events","for","a","given","message",".",":","param","retry",":","(","Optional",")","A","retry","object","used","to","retry","requests",".","If","None","is","specified",",","requests","will","not","be","retried",".",":","param","timeout",":","(","Optional",")","The","amount","of","time",",","in","seconds",",","to","wait","for","the","request","to","complete",".","Note","that","if","retry","is","specified",",","the","timeout","applies","to","each","individual","attempt",".",":","param","metadata",":","(","Optional",")","Additional","metadata","that","is","provided","to","the","method",".",":","return",":","subscription","name","which","will","be","the","system-generated","value","if","the","``","subscription","``","parameter","is","not","supplied"]
code: def create_subscription(
      self,
      topic: str,
      project_id: str = PROVIDE_PROJECT_ID,
      subscription: str | None = None,
      subscription_project_id: str | None = None,
      ack_deadline_secs: int = 10,
      fail_if_exists: bool = False,
      push_config: dict | PushConfig | None = None,
      retain_acked_messages: bool | None = None,
      message_retention_duration: dict | Duration | None = None,
      labels: dict[str, str] | None = None,
      enable_message_ordering: bool = False,
      expiration_policy: dict | ExpirationPolicy | None = None,
      filter_: str | None = None,
      dead_letter_policy: dict | DeadLetterPolicy | None = None,
      retry_policy: dict | RetryPolicy | None = None,
      retry: Retry | _MethodDefault = DEFAULT,
      timeout: float | None = None,
      metadata: Sequence[tuple[str, str]] = (),
  ) -> str:
      """
      Creates a Pub/Sub subscription, if it does not already exist.
  
      :param topic: the Pub/Sub topic name that the subscription will be bound
          to create; do not include the ``projects/{project}/subscriptions/`` prefix.
      :param project_id: Optional, the Google Cloud project ID of the topic that the subscription will be
          bound to. If set to None or missing, the default project_id from the Google Cloud connection
          is used.
      :param subscription: the Pub/Sub subscription name. If empty, a random
          name will be generated using the uuid module
      :param subscription_project_id: the Google Cloud project ID where the subscription
          will be created. If unspecified, ``project_id`` will be used.
      :param ack_deadline_secs: Number of seconds that a subscriber has to
          acknowledge each message pulled from the subscription
      :param fail_if_exists: if set, raise an exception if the topic
          already exists
      :param push_config: If push delivery is used with this subscription,
          this field is used to configure it. An empty ``pushConfig`` signifies
          that the subscriber will pull and ack messages using API methods.
      :param retain_acked_messages: Indicates whether to retain acknowledged
          messages. If true, then messages are not expunged from the subscription's
          backlog, even if they are acknowledged, until they fall out of the
          ``message_retention_duration`` window. This must be true if you would
          like to Seek to a timestamp.
      :param message_retention_duration: How long to retain unacknowledged messages
          in the subscription's backlog, from the moment a message is published. If
          ``retain_acked_messages`` is true, then this also configures the
          retention of acknowledged messages, and thus configures how far back in
          time a ``Seek`` can be done. Defaults to 7 days. Cannot be more than 7
          days or less than 10 minutes.
      :param labels: Client-assigned labels; see
          https://cloud.google.com/pubsub/docs/labels
      :param enable_message_ordering: If true, messages published with the same
          ordering_key in PubsubMessage will be delivered to the subscribers in the order
          in which they are received by the Pub/Sub system. Otherwise, they may be
          delivered in any order.
      :param expiration_policy: A policy that specifies the conditions for this
          subscription's expiration. A subscription is considered active as long as any
          connected subscriber is successfully consuming messages from the subscription or
          is issuing operations on the subscription. If expiration_policy is not set,
          a default policy with ttl of 31 days will be used. The minimum allowed value for
          expiration_policy.ttl is 1 day.
      :param filter_: An expression written in the Cloud Pub/Sub filter language. If
          non-empty, then only PubsubMessages whose attributes field matches the filter are
          delivered on this subscription. If empty, then no messages are filtered out.
      :param dead_letter_policy: A policy that specifies the conditions for dead lettering
          messages in this subscription. If dead_letter_policy is not set, dead lettering is
          disabled.
      :param retry_policy: A policy that specifies how Pub/Sub retries message delivery
          for this subscription. If not set, the default retry policy is applied. This
          generally implies that messages will be retried as soon as possible for healthy
          subscribers. RetryPolicy will be triggered on NACKs or acknowledgement deadline
          exceeded events for a given message.
      :param retry: (Optional) A retry object used to retry requests.
          If None is specified, requests will not be retried.
      :param timeout: (Optional) The amount of time, in seconds, to wait for the request
          to complete. Note that if retry is specified, the timeout applies to each
          individual attempt.
      :param metadata: (Optional) Additional metadata that is provided to the method.
      :return: subscription name which will be the system-generated value if
          the ``subscription`` parameter is not supplied
      """
      subscriber = self.subscriber_client
  
      if not subscription:
          subscription = f"sub-{uuid4()}"
      if not subscription_project_id:
          subscription_project_id = project_id
  
      # Add airflow-version label to the subscription
      labels = labels or {}
      labels["airflow-version"] = "v" + version.replace(
          ".", "-"
      ).replace("+", "-")
  
      subscription_path = f"projects/{subscription_project_id}/subscriptions/{subscription}"
      topic_path = f"projects/{project_id}/topics/{topic}"
  
      self.log.info(
          "Creating subscription (path) %s for topic (path) %a",
          subscription_path,
          topic_path,
      )
      try:
          subscriber.create_subscription(
              request={
                  "name": subscription_path,
                  "topic": topic_path,
                  "push_config": push_config,
                  "ack_deadline_seconds": ack_deadline_secs,
                  "retain_acked_messages": retain_acked_messages,
                  "message_retention_duration": message_retention_duration,
                  "labels": labels,
                  "enable_message_ordering": enable_message_ordering,
                  "expiration_policy": expiration_policy,
                  "filter": filter_,
                  "dead_letter_policy": dead_letter_policy,
                  "retry_policy": retry_policy,
              },
              retry=retry,
              timeout=timeout,
              metadata=metadata,
          )
      except AlreadyExists:
          self.log.warning(
              "Subscription already exists: %s", subscription_path
          )
          if fail_if_exists:
              raise PubSubException(
                  f"Subscription already exists: {subscription_path}"
              )
      except GoogleAPICallError as e:
          raise PubSubException(
              f"Error creating subscription {subscription_path}", e
          )
  
      self.log.info(
          "Created subscription (path) %s for topic (path) %s",
          subscription_path,
          topic_path,
      )
      return subscription
  
imports: from __future__ import annotations
  import warnings
  from base64 import b64decode
  from functools import cached_property
  from typing import TYPE_CHECKING, Any, Sequence
  from uuid import uuid4
  from google.api_core.exceptions import AlreadyExists, GoogleAPICallError
  from google.api_core.gapic_v1.method import DEFAULT, _MethodDefault
  from google.cloud.exceptions import NotFound
  from google.cloud.pubsub_v1 import PublisherClient, SubscriberClient
  from google.pubsub_v1.services.subscriber.async_client import SubscriberAsyncClient
  from googleapiclient.errors import HttpError
  from airflow.exceptions import AirflowProviderDeprecationWarning
  from airflow.providers.google.common.consts import CLIENT_INFO
  from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID, GoogleBaseAsyncHook, GoogleBaseHook
  from airflow.version import version
label_desc: Use image node_id 8 for calling the PubSubHook obj's underlying member method code with example usage: obj.create_subscription(topic, project_id, subscription, subscription_project_id, ack_deadline_secs, fail_if_exists, push_config, retain_acked_messages, message_retention_duration, labels, enable_message_ordering, expiration_policy, filter_, dead_letter_policy, retry_policy, retry, timeout, metadata) and returns: subscription