Make project_id argument optional in all dataproc operators (#21866)

Rafał Harabień · web-flow · commit fb6b2d1fee6a · 2022-03-01T21:03:10.000+01:00
`DataprocHook` handles `project_id=None` in `@GoogleBaseHook.fallback_to_default_project_id` decorator. In some operators `project_id` was already optional. Make it optional for all.
diff --git a/airflow/providers/google/cloud/operators/dataproc.py b/airflow/providers/google/cloud/operators/dataproc.py
@@ -756,9 +756,9 @@ class DataprocDeleteClusterOperator(BaseOperator):
     """
     Deletes a cluster in a project.
 
-    :param project_id: Required. The ID of the Google Cloud project that the cluster belongs to (templated).
     :param region: Required. The Cloud Dataproc region in which to handle the request (templated).
     :param cluster_name: Required. The cluster name (templated).
+    :param project_id: Optional. The ID of the Google Cloud project that the cluster belongs to (templated).
     :param cluster_uuid: Optional. Specifying the ``cluster_uuid`` means the RPC should fail
         if cluster with specified UUID does not exist.
     :param request_id: Optional. A unique id used to identify the request. If the server receives two
@@ -785,9 +785,9 @@ class DataprocDeleteClusterOperator(BaseOperator):
     def __init__(
         self,
         *,
-        project_id: str,
         region: str,
         cluster_name: str,
+        project_id: Optional[str] = None,
         cluster_uuid: Optional[str] = None,
         request_id: Optional[str] = None,
         retry: Optional[Retry] = None,
@@ -1499,7 +1499,7 @@ class DataprocCreateWorkflowTemplateOperator(BaseOperator):
     """
     Creates new workflow template.
 
-    :param project_id: Required. The ID of the Google Cloud project the cluster belongs to.
+    :param project_id: Optional. The ID of the Google Cloud project the cluster belongs to.
     :param region: Required. The Cloud Dataproc region in which to handle the request.
     :param location: (To be deprecated). The Cloud Dataproc region in which to handle the request.
     :param template: The Dataproc workflow template to create. If a dict is provided,
@@ -1519,8 +1519,8 @@ def __init__(
         self,
         *,
         template: Dict,
-        project_id: str,
         region: Optional[str] = None,
+        project_id: Optional[str] = None,
         location: Optional[str] = None,
         retry: Optional[Retry] = None,
         timeout: Optional[float] = None,
@@ -1759,7 +1759,7 @@ class DataprocSubmitJobOperator(BaseOperator):
     """
     Submits a job to a cluster.
 
-    :param project_id: Required. The ID of the Google Cloud project that the job belongs to.
+    :param project_id: Optional. The ID of the Google Cloud project that the job belongs to.
     :param region: Required. The Cloud Dataproc region in which to handle the request.
     :param location: (To be deprecated). The Cloud Dataproc region in which to handle the request.
     :param job: Required. The job resource.
@@ -1798,8 +1798,8 @@ class DataprocSubmitJobOperator(BaseOperator):
     def __init__(
         self,
         *,
-        project_id: str,
         job: Dict,
+        project_id: Optional[str] = None,
         region: Optional[str] = None,
         location: Optional[str] = None,
         request_id: Optional[str] = None,
@@ -1876,8 +1876,8 @@ class DataprocUpdateClusterOperator(BaseOperator):
     """
     Updates a cluster in a project.
 
-    :param project_id: Required. The ID of the Google Cloud project the cluster belongs to.
     :param region: Required. The Cloud Dataproc region in which to handle the request.
+    :param project_id: Optional. The ID of the Google Cloud project the cluster belongs to.
     :param location: (To be deprecated). The Cloud Dataproc region in which to handle the request.
     :param cluster_name: Required. The cluster name.
     :param cluster: Required. The changes to the cluster.
@@ -1993,7 +1993,7 @@ class DataprocCreateBatchOperator(BaseOperator):
     """
     Creates a batch workload.
 
-    :param project_id: Required. The ID of the Google Cloud project that the cluster belongs to. (templated)
+    :param project_id: Optional. The ID of the Google Cloud project that the cluster belongs to. (templated)
     :param region: Required. The Cloud Dataproc region in which to handle the request. (templated)
     :param batch: Required. The batch to create. (templated)
     :param batch_id: Optional. The ID to use for the batch, which will become the final component
@@ -2031,7 +2031,7 @@ def __init__(
         self,
         *,
         region: Optional[str] = None,
-        project_id: str,
+        project_id: Optional[str] = None,
         batch: Union[Dict, Batch],
         batch_id: Optional[str] = None,
         request_id: Optional[str] = None,
@@ -2101,8 +2101,8 @@ class DataprocDeleteBatchOperator(BaseOperator):
     :param batch_id: Required. The ID to use for the batch, which will become the final component
         of the batch's resource name.
         This value must be 4-63 characters. Valid characters are /[a-z][0-9]-/.
-    :param project_id: Required. The ID of the Google Cloud project that the cluster belongs to.
     :param region: Required. The Cloud Dataproc region in which to handle the request.
+    :param project_id: Optional. The ID of the Google Cloud project that the cluster belongs to.
     :param retry: A retry object 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
@@ -2126,7 +2126,7 @@ def __init__(
         *,
         batch_id: str,
         region: str,
-        project_id: str,
+        project_id: Optional[str] = None,
         retry: Optional[Retry] = None,
         timeout: Optional[float] = None,
         metadata: Sequence[Tuple[str, str]] = (),
@@ -2165,8 +2165,8 @@ class DataprocGetBatchOperator(BaseOperator):
     :param batch_id: Required. The ID to use for the batch, which will become the final component
         of the batch's resource name.
         This value must be 4-63 characters. Valid characters are /[a-z][0-9]-/.
-    :param project_id: Required. The ID of the Google Cloud project that the cluster belongs to.
     :param region: Required. The Cloud Dataproc region in which to handle the request.
+    :param project_id: Optional. The ID of the Google Cloud project that the cluster belongs to.
     :param retry: A retry object 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
@@ -2191,7 +2191,7 @@ def __init__(
         *,
         batch_id: str,
         region: str,
-        project_id: str,
+        project_id: Optional[Retry] = None,
         retry: Optional[Retry] = None,
         timeout: Optional[float] = None,
         metadata: Sequence[Tuple[str, str]] = (),
@@ -2230,8 +2230,8 @@ class DataprocListBatchesOperator(BaseOperator):
     """
     Lists batch workloads.
 
-    :param project_id: Required. The ID of the Google Cloud project that the cluster belongs to.
     :param region: Required. The Cloud Dataproc region in which to handle the request.
+    :param project_id: Optional. The ID of the Google Cloud project that the cluster belongs to.
     :param page_size: Optional. The maximum number of batches to return in each response. The service may
         return fewer than this value. The default page size is 20; the maximum page size is 1000.
     :param page_token: Optional. A page token received from a previous ``ListBatches`` call.
diff --git a/airflow/providers/google/cloud/sensors/dataproc.py b/airflow/providers/google/cloud/sensors/dataproc.py
@@ -36,10 +36,10 @@ class DataprocJobSensor(BaseSensorOperator):
     """
     Check for the state of a previously submitted Dataproc job.
 
-    :param project_id: The ID of the google cloud project in which
-        to create the cluster. (templated)
     :param dataproc_job_id: The Dataproc job ID to poll. (templated)
     :param region: Required. The Cloud Dataproc region in which to handle the request. (templated)
+    :param project_id: The ID of the google cloud project in which
+        to create the cluster. (templated)
     :param location: (To be deprecated). The Cloud Dataproc region in which to handle the request. (templated)
     :param gcp_conn_id: The connection ID to use connecting to Google Cloud Platform.
     :param wait_timeout: How many seconds wait for job to be ready.
@@ -51,9 +51,9 @@ class DataprocJobSensor(BaseSensorOperator):
     def __init__(
         self,
         *,
-        project_id: str,
         dataproc_job_id: str,
         region: Optional[str] = None,
+        project_id: Optional[str] = None,
         location: Optional[str] = None,
         gcp_conn_id: str = 'google_cloud_default',
         wait_timeout: Optional[int] = None,
