GCSHook: Add flag to ignore NotFound error on blob deletion

This adds an `ignore_error` boolean to `GCSHook.delete`, which will
suppress any 404 errors if a blob to be deleted doesn't exist.

Author: tuzonghua

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

@@ -705,22 +705,30 @@ def is_older_than(self, bucket_name: str, object_name: str, seconds: int) -> boo
                 return True
         return False
 
-    def delete(self, bucket_name: str, object_name: str) -> None:
+    def delete(self, bucket_name: str, object_name: str, ignore_error: bool = False) -> None:
         """
         Delete an object from the bucket.
 
         :param bucket_name: name of the bucket, where the object resides
         :param object_name: name of the object to delete
+        :param ignore_error: (Optional) whether to ignore NotFound exceptions. Default: False
         """
+        on_error = None
+        if ignore_error:
+            on_error = lambda blob: None
         client = self.get_conn()
         bucket = client.bucket(bucket_name)
         blob = bucket.blob(blob_name=object_name)
-        blob.delete()
-        get_hook_lineage_collector().add_input_asset(
-            context=self, scheme="gs", asset_kwargs={"bucket": bucket.name, "key": blob.name}
-        )
-
-        self.log.info("Blob %s deleted.", object_name)
+        try:
+            bucket.delete_blobs([blob], on_error=on_error)
+            get_hook_lineage_collector().add_input_asset(
+                context=self, scheme="gs", asset_kwargs={"bucket": bucket.name, "key": blob.name}
+            )
+            if not ignore_error:
+                self.log.info("Blob %s deleted.", object_name)
+        except NotFound:
+            self.log.warning("Blob %s in bucket %s does not exist.", blob.name, bucket.name)
+            raise
 
     def get_bucket(self, bucket_name: str) -> storage.Bucket:
         """

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

@@ -528,32 +528,57 @@ def test_rewrite_exposes_lineage(self, mock_service, hook_lineage_collector):
 
     @mock.patch("google.cloud.storage.Bucket")
     @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
-    def test_delete(self, mock_service, mock_bucket):
+    def test_delete(self, mock_service, mock_bucket, caplog):
         test_bucket = "test_bucket"
         test_object = "test_object"
         blob_to_be_deleted = storage.Blob(name=test_object, bucket=mock_bucket)
 
-        get_bucket_method = mock_service.return_value.get_bucket
-        get_blob_method = get_bucket_method.return_value.get_blob
-        delete_method = get_blob_method.return_value.delete
+        bucket_method = mock_service.return_value.bucket
+        blob = bucket_method.return_value.blob
+        delete_method = bucket_method.return_value.delete_blobs
         delete_method.return_value = blob_to_be_deleted
 
-        response = self.gcs_hook.delete(bucket_name=test_bucket, object_name=test_object)
+        with caplog.at_level(logging.INFO):
+            response = self.gcs_hook.delete(bucket_name=test_bucket, object_name=test_object)
         assert response is None
+        bucket_method.assert_called_once_with(test_bucket)
+        blob.assert_called_once_with(blob_name=test_object)
+        delete_method.assert_called_once_with([blob.return_value], on_error=None)
+        assert "Blob test_object deleted" in caplog.text
 
     @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
-    def test_delete_nonexisting_object(self, mock_service):
+    def test_delete_nonexisting_object(self, mock_service, caplog):
         test_bucket = "test_bucket"
         test_object = "test_object"
 
         bucket_method = mock_service.return_value.bucket
         blob = bucket_method.return_value.blob
-        delete_method = blob.return_value.delete
+        delete_method = bucket_method.return_value.delete_blobs
         delete_method.side_effect = NotFound(message="Not Found")
 
-        with pytest.raises(NotFound):
+        with pytest.raises(NotFound), caplog.at_level(logging.INFO):
             self.gcs_hook.delete(bucket_name=test_bucket, object_name=test_object)
 
+        bucket_method.assert_called_once_with(test_bucket)
+        blob.assert_called_once_with(blob_name=test_object)
+        delete_method.assert_called_once_with([blob.return_value], on_error=None)
+        assert "does not exist" in caplog.text
+
+    @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
+    def test_delete_nonexisting_object_ignore_error(self, mock_service, caplog):
+        test_bucket = "test_bucket"
+        test_object = "test_object"
+
+        bucket_method = mock_service.return_value.bucket
+        blob = bucket_method.return_value.blob
+        delete_method = bucket_method.return_value.delete_blobs
+
+        self.gcs_hook.delete(bucket_name=test_bucket, object_name=test_object, ignore_error=True)
+
+        bucket_method.assert_called_once_with(test_bucket)
+        blob.assert_called_once_with(blob_name=test_object)
+        delete_method.assert_called_once_with([blob.return_value], on_error=mock.ANY)
+
     @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
     def test_delete_exposes_lineage(self, mock_service, hook_lineage_collector):
         test_bucket = "test_bucket"