class_docstr: Google Kubernetes Engine cluster APIs.
  
  All the methods in the hook where project_id is used must be called with
  keyword arguments rather than positional.
class_docstr_tok: ["Google","Kubernetes","Engine","cluster","APIs",".","All","the","methods","in","the","hook","where","project_id","is","used","must","be","called","with","keyword","arguments","rather","than","positional","."]
docstr: Create a cluster.
  
  This should consist of the specified number, and the type of Google
  Compute Engine instances.
  
  :param cluster: A Cluster protobuf or dict. If dict is provided, it must
      be of the same form as the protobuf message
      :class:`google.cloud.container_v1.types.Cluster`.
  :param project_id: Google Cloud project ID.
  :param wait_to_complete: A boolean value which makes method to sleep
      while operation of creation is not finished.
  :param retry: A retry object (``google.api_core.retry.Retry``) used to
      retry requests. If None is specified, requests will not be retried.
  :param timeout: 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.
  :return: The full url to the new, or existing, cluster.
  :raises ParseError: On JSON parsing problems when trying to convert
      dict.
  :raises AirflowException: cluster is not dict type nor Cluster proto
      type.
docstr_tok: ["Create","a","cluster",".","This","should","consist","of","the","specified","number",",","and","the","type","of","Google","Compute","Engine","instances",".",":","param","cluster",":","A","Cluster","protobuf","or","dict",".","If","dict","is","provided",",","it","must","be","of","the","same","form","as","the","protobuf","message",":","class",":","`","google.cloud.container_v1.types.Cluster","`",".",":","param","project_id",":","Google","Cloud","project","ID",".",":","param","wait_to_complete",":","A","boolean","value","which","makes","method","to","sleep","while","operation","of","creation","is","not","finished",".",":","param","retry",":","A","retry","object","(","``","google.api_core.retry.Retry","``",")","used","to","retry","requests",".","If","None","is","specified",",","requests","will","not","be","retried",".",":","param","timeout",":","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",".",":","return",":","The","full","url","to","the","new",",","or","existing",",","cluster",".",":","raises","ParseError",":","On","JSON","parsing","problems","when","trying","to","convert","dict",".",":","raises","AirflowException",":","cluster","is","not","dict","type","nor","Cluster","proto","type","."]
code: def create_cluster(
      self,
      cluster: dict | Cluster,
      project_id: str = PROVIDE_PROJECT_ID,
      wait_to_complete: bool = True,
      retry: Retry | _MethodDefault = DEFAULT,
      timeout: float | None = None,
  ) -> Operation | Cluster:
      """Create a cluster.
  
      This should consist of the specified number, and the type of Google
      Compute Engine instances.
  
      :param cluster: A Cluster protobuf or dict. If dict is provided, it must
          be of the same form as the protobuf message
          :class:`google.cloud.container_v1.types.Cluster`.
      :param project_id: Google Cloud project ID.
      :param wait_to_complete: A boolean value which makes method to sleep
          while operation of creation is not finished.
      :param retry: A retry object (``google.api_core.retry.Retry``) used to
          retry requests. If None is specified, requests will not be retried.
      :param timeout: 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.
      :return: The full url to the new, or existing, cluster.
      :raises ParseError: On JSON parsing problems when trying to convert
          dict.
      :raises AirflowException: cluster is not dict type nor Cluster proto
          type.
      """
      if isinstance(cluster, dict):
          cluster = Cluster.from_json(json.dumps(cluster))
      elif not isinstance(cluster, Cluster):
          raise AirflowException(
              "cluster is not instance of Cluster proto or python dict"
          )
  
      self._append_label(cluster, "airflow-version", "v" + version.version)  # type: ignore
  
      self.log.info(
          "Creating (project_id=%s, location=%s, cluster_name=%s)",
          project_id,
          self.location,
          cluster.name,  # type: ignore
      )
      operation = self.get_cluster_manager_client().create_cluster(
          parent=f"projects/{project_id}/locations/{self.location}",
          cluster=cluster,  # type: ignore
          retry=retry,
          timeout=timeout,
      )
  
      if wait_to_complete:
          operation = self.wait_for_operation(operation, project_id)
  
      return operation
  
imports: from __future__ import annotations
  import contextlib
  import json
  import time
  import warnings
  from functools import cached_property
  from typing import TYPE_CHECKING, Sequence
  from gcloud.aio.auth import Token
  from google.api_core.exceptions import NotFound
  from google.api_core.gapic_v1.method import DEFAULT, _MethodDefault
  from google.auth.transport import requests
  from google.cloud import container_v1, exceptions
  from google.cloud.container_v1 import ClusterManagerAsyncClient, ClusterManagerClient
  from google.cloud.container_v1.types import Cluster, Operation
  from kubernetes import client
  from kubernetes_asyncio import client
  from kubernetes_asyncio.config.kube_config import FileOrData
  from urllib3.exceptions import HTTPError
  from airflow import version
  from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
  from airflow.providers.cncf.kubernetes.utils.pod_manager import PodOperatorHookProtocol
  from airflow.providers.google.common.consts import CLIENT_INFO
  from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID, GoogleBaseAsyncHook, GoogleBaseHook
label_desc: Use image node_id 9 for calling the GKEHook obj's underlying member method code with example usage: obj.create_cluster(cluster, project_id, wait_to_complete, retry, timeout) and returns: operation