Fix exists method to support using Requester Pays

providers/google/src/airflow/providers/google/cloud/hooks/gcs.py

@@ -598,17 +598,25 @@ def _call_with_retry(f: Callable[[], None]) -> None:
             context=self, scheme="gs", asset_kwargs={"bucket": bucket.name, "key": blob.name}
         )
 
-    def exists(self, bucket_name: str, object_name: str, retry: Retry = DEFAULT_RETRY) -> bool:
+    def exists(
+        self,
+        bucket_name: str,
+        object_name: str,
+        retry: Retry = DEFAULT_RETRY,
+        user_project: str | None = None,
+    ) -> bool:
         """
         Check for the existence of a file in Google Cloud Storage.
 
         :param bucket_name: The Google Cloud Storage bucket where the object is.
         :param object_name: The name of the blob_name to check in the Google cloud
             storage bucket.
         :param retry: (Optional) How to retry the RPC
+        :param user_project: The identifier of the Google Cloud project to bill for the request.
+            Required for Requester Pays buckets.
         """
         client = self.get_conn()
-        bucket = client.bucket(bucket_name)
+        bucket = client.bucket(bucket_name, user_project=user_project)
         blob = bucket.blob(blob_name=object_name)
         return blob.exists(retry=retry)
 
@@ -625,7 +633,7 @@ def get_blob_update_time(self, bucket_name: str, object_name: str):
 
     def is_updated_after(self, bucket_name: str, object_name: str, ts: datetime) -> bool:
         """
-        Check if an blob_name is updated in Google Cloud Storage.
+        Check if a blob_name is updated in Google Cloud Storage.
 
         :param bucket_name: The Google Cloud Storage bucket where the object is.
         :param object_name: The name of the object to check in the Google cloud
@@ -645,7 +653,7 @@ def is_updated_between(
         self, bucket_name: str, object_name: str, min_ts: datetime, max_ts: datetime
     ) -> bool:
         """
-        Check if an blob_name is updated in Google Cloud Storage.
+        Check if a blob_name is updated in Google Cloud Storage.
 
         :param bucket_name: The Google Cloud Storage bucket where the object is.
         :param object_name: The name of the object to check in the Google cloud
@@ -666,7 +674,7 @@ def is_updated_between(
 
     def is_updated_before(self, bucket_name: str, object_name: str, ts: datetime) -> bool:
         """
-        Check if an blob_name is updated before given time in Google Cloud Storage.
+        Check if a blob_name is updated before given time in Google Cloud Storage.
 
         :param bucket_name: The Google Cloud Storage bucket where the object is.
         :param object_name: The name of the object to check in the Google cloud

providers/google/tests/unit/google/cloud/hooks/test_gcs.py

@@ -206,11 +206,11 @@ def test_exists(self, mock_service):
         exists_method.return_value = True
 
         # When
-        response = self.gcs_hook.exists(bucket_name=test_bucket, object_name=test_object)
+        response = self.gcs_hook.exists(bucket_name=test_bucket, object_name=test_object, user_project=None)
 
         # Then
         assert response
-        bucket_mock.assert_called_once_with(test_bucket)
+        bucket_mock.assert_called_once_with(test_bucket, user_project=None)
         blob_object.assert_called_once_with(blob_name=test_object)
         exists_method.assert_called_once_with(retry=DEFAULT_RETRY)
 
@@ -226,7 +226,7 @@ def test_exists_nonexisting_object(self, mock_service):
         exists_method.return_value = False
 
         # When
-        response = self.gcs_hook.exists(bucket_name=test_bucket, object_name=test_object)
+        response = self.gcs_hook.exists(bucket_name=test_bucket, object_name=test_object, user_project=None)
 
         # Then
         assert not response