Batching APIs Raise on Any Failure

sdk/storage/azure-storage-blob/HISTORY.md

@@ -24,6 +24,7 @@
 - `Logging` has been renamed to `BlobAnalyticsLogging`.
 - All operations that take Etag conditional parameters (`if_match` and `if_none_match`) now take explicit `etag` and `match_condition` parameters, where `etag` is the Etag value, and `match_condition` is an instance of `azure.core.MatchConditions`.
 - The `generate_shared_access_signature` methods on each of `BlobServiceClient`, `ContainerClient` and `BlobClient` have been replaced by module level functions `generate_account_sas`, `generate_container_sas` and `generate_blob_sas`.
+- The batch APIs now have an additional keyword only argument `raise_on_any_failure` which defaults to True. This will raise an error even if there's a partial batch failure.
 
 **New features**
 

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

@@ -15,6 +15,7 @@
 from .download import StorageStreamDownloader
 from ._shared_access_signature import generate_account_sas, generate_container_sas, generate_blob_sas
 from ._shared.policies import ExponentialRetry, LinearRetry
+from ._shared.response_handlers import PartialBatchErrorException
 from ._shared.models import(
     LocationMode,
     ResourceTypes,
@@ -203,5 +204,6 @@ def download_blob_from_url(
     'RehydratePriority',
     'generate_account_sas',
     'generate_container_sas',
-    'generate_blob_sas'
+    'generate_blob_sas',
+    'PartialBatchErrorException'
 ]

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py

@@ -53,7 +53,7 @@
     ExponentialRetry,
 )
 from .._generated.models import StorageErrorException
-from .response_handlers import process_storage_error
+from .response_handlers import process_storage_error, PartialBatchErrorException
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -197,6 +197,8 @@ def _batch_send(
     ):
         """Given a series of request, do a Storage batch call.
         """
+        # Pop it here, so requests doesn't feel bad about additional kwarg
+        raise_on_any_failure = kwargs.pop("raise_on_any_failure", True)
         request = self._client._client.post(  # pylint: disable=protected-access
             url='https://{}/?comp=batch'.format(self.primary_hostname),
             headers={
@@ -220,7 +222,17 @@ def _batch_send(
         try:
             if response.status_code not in [202]:
                 raise HttpResponseError(response=response)
-            return response.parts()
+            parts = response.parts()
+            if raise_on_any_failure:
+                parts = list(response.parts())
+                if any(p for p in parts if not 200 <= p.status_code < 300):
+                    error = PartialBatchErrorException(
+                        message="There is a partial failure in the batch operation.",
+                        response=response, parts=parts
+                    )
+                    raise error
+                return iter(parts)
+            return parts
         except StorageErrorException as error:
             process_storage_error(error)
 

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py

@@ -9,8 +9,8 @@
     TYPE_CHECKING
 )
 import logging
-
 from azure.core.pipeline import AsyncPipeline
+from azure.core.async_paging import AsyncList
 from azure.core.exceptions import HttpResponseError
 from azure.core.pipeline.policies import (
     ContentDecodePolicy,
@@ -33,7 +33,7 @@
 from .policies_async import AsyncStorageResponseHook
 
 from .._generated.models import StorageErrorException
-from .response_handlers import process_storage_error
+from .response_handlers import process_storage_error, PartialBatchErrorException
 
 if TYPE_CHECKING:
     from azure.core.pipeline import Pipeline
@@ -101,6 +101,8 @@ async def _batch_send(
     ):
         """Given a series of request, do a Storage batch call.
         """
+        # Pop it here, so requests doesn't feel bad about additional kwarg
+        raise_on_any_failure = kwargs.pop("raise_on_any_failure", True)
         request = self._client._client.post(  # pylint: disable=protected-access
             url='https://{}/?comp=batch'.format(self.primary_hostname),
             headers={
@@ -124,7 +126,19 @@ async def _batch_send(
         try:
             if response.status_code not in [202]:
                 raise HttpResponseError(response=response)
-            return response.parts()  # Return an AsyncIterator
+            parts = response.parts() # Return an AsyncIterator
+            if raise_on_any_failure:
+                parts_list = []
+                async for part in parts:
+                    parts_list.append(part)
+                if any(p for p in parts_list if not 200 <= p.status_code < 300):
+                    error = PartialBatchErrorException(
+                        message="There is a partial failure in the batch operation.",
+                        response=response, parts=parts_list
+                    )
+                    raise error
+                return AsyncList(parts_list)
+            return parts
         except StorageErrorException as error:
             process_storage_error(error)
 

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/response_handlers.py

@@ -31,6 +31,19 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+class PartialBatchErrorException(HttpResponseError):
+    """There is a partial failure in batch operations.
+
+    :param message: The message of the exception.
+    :param response: Server response to be deserialized.
+    :param parts: A list of the parts in multipart response.
+    """
+
+    def __init__(self, message, response, parts):
+        self.parts = parts
+        super(PartialBatchErrorException, self).__init__(message=message, response=response)
+
+
 def parse_length_from_content_range(content_range):
     '''
     Parses the blob length from the content range header: bytes 1-3/65537

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

@@ -800,6 +800,10 @@ async def delete_blobs(  # pylint: disable=arguments-differ
             If a date is passed in without timezone info, it is assumed to be UTC.
             Specify this header to perform the operation only if
             the resource has not been modified since the specified date/time.
+        :keyword bool raise_on_any_failure:
+            This is a boolean param which defaults to True. When this is set, an exception
+            is raised even if there is a single operation failure. For optimal performance,
+            this should be set to False
         :keyword str etag:
             An ETag value, or the wildcard character (*). Used to check if the resource has changed,
             and act according to the condition specified by the `match_condition` parameter.
@@ -810,13 +814,15 @@ async def delete_blobs(  # pylint: disable=arguments-differ
         :return: An async iterator of responses, one for each blob in order
         :rtype: asynciterator[~azure.core.pipeline.transport.AsyncHttpResponse]
         """
+        raise_on_any_failure = kwargs.pop('raise_on_any_failure', True)
         timeout = kwargs.pop('timeout', None)
         options = BlobClient._generic_delete_blob_options(  # pylint: disable=protected-access
             delete_snapshots=delete_snapshots,
             lease=lease,
             timeout=timeout,
             **kwargs
         )
+        options.update({'raise_on_any_failure': raise_on_any_failure})
         query_parameters, header_parameters = self._generate_delete_blobs_options(**options)
         # To pass kwargs to "_batch_send", we need to remove anything that was
         # in the Autorest signature for Autorest, otherwise transport will be upset
@@ -863,6 +869,10 @@ async def set_standard_blob_tier_blobs(
             Required if the blob has an active lease. Value can be a LeaseClient object
             or the lease ID as a string.
         :type lease: ~azure.storage.blob.lease.LeaseClient or str
+        :keyword bool raise_on_any_failure:
+            This is a boolean param which defaults to True. When this is set, an exception
+            is raised even if there is a single operation failure. For optimal performance,
+            this should be set to False.
         :return: An async iterator of responses, one for each blob in order
         :rtype: asynciterator[~azure.core.pipeline.transport.AsyncHttpResponse]
         """
@@ -916,6 +926,10 @@ async def set_premium_page_blob_tier_blobs(
             Required if the blob has an active lease. Value can be a LeaseClient object
             or the lease ID as a string.
         :type lease: ~azure.storage.blob.lease.LeaseClient or str
+        :keyword bool raise_on_any_failure:
+            This is a boolean param which defaults to True. When this is set, an exception
+            is raised even if there is a single operation failure. For optimal performance,
+            this should be set to False.
         :return: An async iterator of responses, one for each blob in order
         :rtype: asynciterator[~azure.core.pipeline.transport.AsyncHttpResponse]
         """

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

@@ -952,6 +952,9 @@ def delete_blobs(self, *blobs, **kwargs):
             If a date is passed in without timezone info, it is assumed to be UTC.
             Specify this header to perform the operation only if
             the resource has not been modified since the specified date/time.
+        :keyword bool raise_on_any_failure:
+            This is a boolean param which defaults to True. When this is set, an exception
+            is raised even if there is a single operation failure.
         :keyword str etag:
             An ETag value, or the wildcard character (*). Used to check if the resource has changed,
             and act according to the condition specified by the `match_condition` parameter.
@@ -962,9 +965,11 @@ def delete_blobs(self, *blobs, **kwargs):
         :return: An iterator of responses, one for each blob in order
         :rtype: iterator[~azure.core.pipeline.transport.HttpResponse]
         """
+        raise_on_any_failure = kwargs.pop('raise_on_any_failure', True)
         options = BlobClient._generic_delete_blob_options(  # pylint: disable=protected-access
             **kwargs
         )
+        options.update({'raise_on_any_failure': raise_on_any_failure})
         query_parameters, header_parameters = self._generate_delete_blobs_options(**options)
         # To pass kwargs to "_batch_send", we need to remove anything that was
         # in the Autorest signature for Autorest, otherwise transport will be upset
@@ -1045,6 +1050,9 @@ def set_standard_blob_tier_blobs(
             Required if the blob has an active lease. Value can be a LeaseClient object
             or the lease ID as a string.
         :type lease: ~azure.storage.blob.LeaseClient or str
+        :keyword bool raise_on_any_failure:
+            This is a boolean param which defaults to True. When this is set, an exception
+            is raised even if there is a single operation failure.
         :return: An iterator of responses, one for each blob in order
         :rtype: iterator[~azure.core.pipeline.transport.HttpResponse]
         """
@@ -1099,6 +1107,9 @@ def set_premium_page_blob_tier_blobs(
             Required if the blob has an active lease. Value can be a LeaseClient object
             or the lease ID as a string.
         :type lease: ~azure.storage.blob.LeaseClient or str
+        :keyword bool raise_on_any_failure:
+            This is a boolean param which defaults to True. When this is set, an exception
+            is raised even if there is a single operation failure.
         :return: An iterator of responses, one for each blob in order
         :rtype: iterator[~azure.core.pipeline.transport.HttpResponse]
         """