Added blob exists method (#13221)

* added feature and unit tests

* fixed failing test issue

* added async method and more unit tests

* ffixed passed parameters

* fixed python 27 issue with kwargs

* reset commit

* Update _blob_client_async.py

removed unused import, fixed linter

* Update sdk/storage/azure-storage-blob/azure/storage/blob/aio/_blob_client_async.py

Co-authored-by: Xiaoxi Fu <49707495+xiafu-msft@users.noreply.github.com>

* Update sdk/storage/azure-storage-blob/azure/storage/blob/_blob_client.py

Co-authored-by: Xiaoxi Fu <49707495+xiafu-msft@users.noreply.github.com>

* fixed failing tests

* fixed linting/import order

Co-authored-by: Xiaoxi Fu <49707495+xiafu-msft@users.noreply.github.com>

sdk/storage/azure-storage-blob/azure/storage/blob/_blob_client.py

@@ -18,6 +18,7 @@
 
 import six
 from azure.core.tracing.decorator import distributed_trace
+from azure.core.exceptions import ResourceNotFoundError
 
 from ._shared import encode_base64
 from ._shared.base_client import StorageAccountHostsMixin, parse_connection_str, parse_query
@@ -929,6 +930,31 @@ def undelete_blob(self, **kwargs):
         except StorageErrorException as error:
             process_storage_error(error)
 
+    @distributed_trace()
+    def exists(self, **kwargs):
+        # type: (**Any) -> bool
+        """
+        Returns True if a blob exists with the defined parameters, and returns
+        False otherwise.
+
+        :param str version_id:
+            The version id parameter is an opaque DateTime
+            value that, when present, specifies the version of the blob to check if it exists.
+        :param int timeout:
+            The timeout parameter is expressed in seconds.
+        :returns: boolean
+        """
+        try:
+            self._client.blob.get_properties(
+                snapshot=self.snapshot,
+                **kwargs)
+            return True
+        except StorageErrorException as error:
+            try:
+                process_storage_error(error)
+            except ResourceNotFoundError:
+                return False
+
     @distributed_trace
     def get_blob_properties(self, **kwargs):
         # type: (**Any) -> BlobProperties

sdk/storage/azure-storage-blob/azure/storage/blob/aio/_blob_client_async.py

@@ -11,6 +11,7 @@
 )
 
 from azure.core.tracing.decorator_async import distributed_trace_async
+from azure.core.exceptions import ResourceNotFoundError
 
 from .._shared.base_client_async import AsyncStorageAccountHostsMixin
 from .._shared.policies_async import ExponentialRetry
@@ -30,6 +31,7 @@
 from ._lease_async import BlobLeaseClient
 from ._download_async import StorageStreamDownloader
 
+
 if TYPE_CHECKING:
     from datetime import datetime
     from .._models import (  # pylint: disable=unused-import
@@ -459,6 +461,31 @@ async def undelete_blob(self, **kwargs):
         except StorageErrorException as error:
             process_storage_error(error)
 
+    @distributed_trace_async
+    async def exists(self, **kwargs):
+        # type: (**Any) -> bool
+        """
+        Returns True if a blob exists with the defined parameters, and returns
+        False otherwise.
+
+        :param str version_id:
+            The version id parameter is an opaque DateTime
+            value that, when present, specifies the version of the blob to check if it exists.
+        :param int timeout:
+            The timeout parameter is expressed in seconds.
+        :returns: boolean
+        """
+        try:
+            await self._client.blob.get_properties(
+                snapshot=self.snapshot,
+                **kwargs)
+            return True
+        except StorageErrorException as error:
+            try:
+                process_storage_error(error)
+            except ResourceNotFoundError:
+                return False
+
     @distributed_trace_async
     async def get_blob_properties(self, **kwargs):
         # type: (Any) -> BlobProperties