Updated blobs shared code

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

@@ -100,6 +100,9 @@ def _add_authorization_header(self, request, string_to_sign):
             raise _wrap_exception(ex, AzureSigningError)
 
     def on_request(self, request, **kwargs):
+        if not 'content-type' in request.http_request.headers:
+            request.http_request.headers['content-type'] = 'application/xml; charset=utf-8'
+
         string_to_sign = \
             self._get_verb(request) + \
             self._get_headers(

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

@@ -87,9 +87,7 @@ def __init__(
         self.require_encryption = kwargs.get('require_encryption', False)
         self.key_encryption_key = kwargs.get('key_encryption_key')
         self.key_resolver_function = kwargs.get('key_resolver_function')
-
-        self._config, self._pipeline = create_pipeline(
-            self.credential, storage_sdk=service, hosts=self._hosts, **kwargs)
+        self._config, self._pipeline = self._create_pipeline(self.credential, storage_sdk=service, **kwargs)
 
     def __enter__(self):
         self._client.__enter__()
@@ -145,6 +143,38 @@ def _format_query_string(self, sas_token, credential, snapshot=None, share_snaps
             credential = None
         return query_str.rstrip('?&'), credential
 
+    def _create_pipeline(self, credential, **kwargs):
+        # type: (Any, **Any) -> Tuple[Configuration, Pipeline]
+        credential_policy = None
+        if hasattr(credential, 'get_token'):
+            credential_policy = BearerTokenCredentialPolicy(credential, STORAGE_OAUTH_SCOPE)
+        elif isinstance(credential, SharedKeyCredentialPolicy):
+            credential_policy = credential
+        elif credential is not None:
+            raise TypeError("Unsupported credential: {}".format(credential))
+
+        config = kwargs.get('_configuration') or create_configuration(**kwargs)
+        if kwargs.get('_pipeline'):
+            return config, kwargs['_pipeline']
+        config.transport = kwargs.get('transport')  # type: HttpTransport
+        if not config.transport:
+            config.transport = RequestsTransport(config)
+        policies = [
+            QueueMessagePolicy(),
+            config.headers_policy,
+            config.user_agent_policy,
+            StorageContentValidation(),
+            StorageRequestHook(**kwargs),
+            credential_policy,
+            ContentDecodePolicy(),
+            RedirectPolicy(**kwargs),
+            StorageHosts(hosts=self._hosts, **kwargs),
+            config.retry_policy,
+            config.logging_policy,
+            StorageResponseHook(**kwargs),
+        ]
+        return config, Pipeline(config.transport, policies=policies)
+
 
 def format_shared_key_credential(account, credential):
     if isinstance(credential, six.string_types):
@@ -219,7 +249,6 @@ def create_configuration(**kwargs):
     config.headers_policy = StorageHeadersPolicy(**kwargs)
     config.user_agent_policy = StorageUserAgentPolicy(**kwargs)
     config.retry_policy = kwargs.get('retry_policy') or ExponentialRetry(**kwargs)
-    config.redirect_policy = RedirectPolicy(**kwargs)
     config.logging_policy = StorageLoggingPolicy(**kwargs)
     config.proxy_policy = ProxyPolicy(**kwargs)
 
@@ -244,39 +273,6 @@ def create_configuration(**kwargs):
     return config
 
 
-def create_pipeline(credential, **kwargs):
-    # type: (Any, **Any) -> Tuple[Configuration, Pipeline]
-    credential_policy = None
-    if hasattr(credential, 'get_token'):
-        credential_policy = BearerTokenCredentialPolicy(credential, STORAGE_OAUTH_SCOPE)
-    elif isinstance(credential, SharedKeyCredentialPolicy):
-        credential_policy = credential
-    elif credential is not None:
-        raise TypeError("Unsupported credential: {}".format(credential))
-
-    config = kwargs.get('_configuration') or create_configuration(**kwargs)
-    if kwargs.get('_pipeline'):
-        return config, kwargs['_pipeline']
-    transport = kwargs.get('transport')  # type: HttpTransport
-    if not transport:
-        transport = RequestsTransport(config)
-    policies = [
-        QueueMessagePolicy(),
-        config.headers_policy,
-        config.user_agent_policy,
-        StorageContentValidation(),
-        StorageRequestHook(**kwargs),
-        credential_policy,
-        ContentDecodePolicy(),
-        config.redirect_policy,
-        StorageHosts(**kwargs),
-        config.retry_policy,
-        config.logging_policy,
-        StorageResponseHook(**kwargs),
-    ]
-    return config, Pipeline(transport, policies=policies)
-
-
 def parse_query(query_str):
     sas_values = QueryStringConstants.to_list()
     parsed_query = {k: v[0] for k, v in parse_qs(query_str).items()}

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

@@ -0,0 +1,84 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+from typing import (  # pylint: disable=unused-import
+    Union, Optional, Any, Iterable, Dict, List, Type, Tuple,
+    TYPE_CHECKING
+)
+import logging
+
+from azure.core.pipeline import AsyncPipeline
+try:
+    from azure.core.pipeline.transport import AioHttpTransport as AsyncTransport
+except ImportError:
+    from azure.core.pipeline.transport import AsyncioRequestsTransport as AsyncTransport
+from azure.core.pipeline.policies import (
+    ContentDecodePolicy,
+    BearerTokenCredentialPolicy,
+    AsyncRedirectPolicy)
+
+from .constants import STORAGE_OAUTH_SCOPE, DEFAULT_SOCKET_TIMEOUT
+from .authentication import SharedKeyCredentialPolicy
+from .base_client import create_configuration
+from .policies import (
+    StorageContentValidation,
+    StorageRequestHook,
+    StorageHosts,
+    QueueMessagePolicy)
+from .policies_async import AsyncStorageResponseHook
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AsyncStorageAccountHostsMixin(object):
+
+    def __enter__(self):
+        raise TypeError("Async client only supports 'async with'.")
+
+    def __exit__(self, *args):
+        pass
+
+    async def __aenter__(self):
+        await self._client.__aenter__()
+        return self
+
+    async def __aexit__(self, *args):
+        await self._client.__aexit__(*args)
+
+    def _create_pipeline(self, credential, **kwargs):
+        # type: (Any, **Any) -> Tuple[Configuration, Pipeline]
+        credential_policy = None
+        if hasattr(credential, 'get_token'):
+            credential_policy = BearerTokenCredentialPolicy(credential, STORAGE_OAUTH_SCOPE)
+        elif isinstance(credential, SharedKeyCredentialPolicy):
+            credential_policy = credential
+        elif credential is not None:
+            raise TypeError("Unsupported credential: {}".format(credential))
+
+        if 'connection_timeout' not in kwargs:
+            kwargs['connection_timeout'] = DEFAULT_SOCKET_TIMEOUT[0]
+        config = kwargs.get('_configuration') or create_configuration(**kwargs)
+        if kwargs.get('_pipeline'):
+            return config, kwargs['_pipeline']
+        config.transport = kwargs.get('transport')  # type: HttpTransport
+        if not config.transport:
+            config.transport = AsyncTransport(config)
+        policies = [
+            QueueMessagePolicy(),
+            config.headers_policy,
+            config.user_agent_policy,
+            StorageContentValidation(),
+            StorageRequestHook(**kwargs),
+            credential_policy,
+            ContentDecodePolicy(),
+            AsyncRedirectPolicy(**kwargs),
+            StorageHosts(hosts=self._hosts, **kwargs),
+            config.retry_policy,
+            config.logging_policy,
+            AsyncStorageResponseHook(**kwargs),
+        ]
+        return config, AsyncPipeline(config.transport, policies=policies)

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/download_chunking.py → sdk/storage/azure-storage-blob/azure/storage/blob/_shared/downloads.py

@@ -10,7 +10,6 @@
 
 from azure.core.exceptions import HttpResponseError
 
-from .models import ModifiedAccessConditions
 from .request_handlers import validate_and_format_range_headers
 from .response_handlers import process_storage_error, parse_length_from_content_range
 from .encryption import decrypt_blob
@@ -40,22 +39,25 @@ def process_range_and_offset(start_range, end_range, length, encryption):
 
 
 def process_content(data, start_offset, end_offset, encryption):
-    if encryption.get('key') is not None or encryption.get('resolver') is not None:
+    if data is None:
+        raise ValueError("Response cannot be None.")
+    content = b"".join(list(data))
+    if content and encryption.get('key') is not None or encryption.get('resolver') is not None:
         try:
             return decrypt_blob(
                 encryption.get('required'),
                 encryption.get('key'),
                 encryption.get('resolver'),
-                data,
+                content,
                 start_offset,
-                end_offset)
+                end_offset,
+                data.response.headers)
         except Exception as error:
             raise HttpResponseError(
                 message="Decryption failed.",
                 response=data.response,
                 error=error)
-    else:
-        return b"".join(list(data))
+    return content
 
 
 class _ChunkDownloader(object):
@@ -209,7 +211,7 @@ def _write_to_stream(self, chunk_data, chunk_start):
         self.stream.write(chunk_data)
 
 
-class StorageStreamDownloader(object):
+class StorageStreamDownloader(object):  # pylint: disable=too-many-instance-attributes
     """A streaming object to download from Azure Storage.
 
     The stream downloader can iterated, or download to open file or stream
@@ -291,14 +293,14 @@ def __iter__(self):
             # Use the length unless it is over the end of the file
             data_end = min(self.file_size, self.length + 1)
 
-        downloader = SequentialBlobChunkDownloader(
+        downloader = SequentialChunkDownloader(
             service=self.service,
             total_size=self.download_size,
             chunk_size=self.config.max_chunk_get_size,
             current_progress=self.first_get_size,
             start_range=self.initial_range[1] + 1,  # start where the first download ended
             end_range=data_end,
-            stream=stream,
+            stream=None,
             validate_content=self.validate_content,
             encryption_options=self.encryption_options,
             use_location=self.location_mode,