use blob.exists to check the GCS file (#34818)

* use blob.exists to check the GCS file

* added try

* fixed the tests

Author: liferoad

sdks/python/apache_beam/io/gcp/gcsio.py

@@ -481,11 +481,10 @@ def exists(self, path):
     Args:
       path: GCS file path pattern in the form gs://<bucket>/<name>.
     """
-    try:
-      self._gcs_object(path)
-      return True
-    except NotFound:
-      return False
+    bucket_name, blob_name = parse_gcs_path(path)
+    bucket = self.client.bucket(bucket_name)
+    blob = bucket.blob(blob_name)
+    return blob.exists(retry=self._storage_client_retry)
 
   def checksum(self, path):
     """Looks up the checksum of a GCS object.

sdks/python/apache_beam/io/gcp/gcsio_test.py

@@ -231,6 +231,9 @@ def download_as_bytes(self, **kwargs):
   def __eq__(self, other):
     return self.bucket.get_blob(self.name) is other.bucket.get_blob(other.name)
 
+  def exists(self, **kwargs):
+    return self.bucket.get_blob(self.name) is not None
+
 
 @unittest.skipIf(NotFound is None, 'GCP dependencies are not installed')
 class TestGCSPathParser(unittest.TestCase):
@@ -396,7 +399,7 @@ def test_exists(self):
     self.assertFalse(self.gcs.exists(file_name + 'xyz'))
     self.assertTrue(self.gcs.exists(file_name))
 
-  @mock.patch.object(FakeBucket, 'get_blob')
+  @mock.patch.object(FakeBlob, 'exists')
   def test_exists_failure(self, mock_get):
     # Raising an error other than 404. Raising 404 is a valid failure for
     # exists() call.