Refactored sync blobs

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

@@ -34,7 +34,7 @@ def _parallel_uploads(executor, uploader, pending, running):
         except StopIteration:
             break
         else:
-            running.add(executor.submit(uploader.process_chunk, next_chunk))
+            running.add(executor.submit(uploader, next_chunk))
 
     # Wait for the remaining uploads to finish
     done, _running = futures.wait(running)
@@ -55,7 +55,7 @@ def upload_data_chunks(
 
     if encryption_options:
         encryptor, padder = get_blob_encryptor_and_padder(
-            encryption_options.get('key'),
+            encryption_options.get('cek'),
             encryption_options.get('vector'),
             uploader_class is not PageBlobChunkUploader)
         kwargs['encryptor'] = encryptor
@@ -74,20 +74,18 @@ def upload_data_chunks(
         parallel=parallel,
         validate_content=validate_content,
         **kwargs)
-
     if parallel:
         executor = futures.ThreadPoolExecutor(max_connections)
         upload_tasks = uploader.get_chunk_streams()
         running_futures = [
             executor.submit(uploader.process_chunk, u)
             for u in islice(upload_tasks, 0, max_connections)
         ]
-        range_ids = _parallel_uploads(executor, uploader, upload_tasks, running_futures)
+        range_ids = _parallel_uploads(executor, uploader.process_chunk, upload_tasks, running_futures)
     else:
         range_ids = [uploader.process_chunk(result) for result in uploader.get_chunk_streams()]
-
     if any(range_ids):
-        return range_ids
+        return [r[1] for r in sorted(range_ids, key=lambda r: r[0])]
     return uploader.response_headers
 
 
@@ -118,8 +116,10 @@ def upload_substream_blocks(
             executor.submit(uploader.process_substream_block, u)
             for u in islice(upload_tasks, 0, max_connections)
         ]
-        return _parallel_uploads(executor, uploader, upload_tasks, running_futures)
-    return [uploader.process_substream_block(b) for b in uploader.get_substream_blocks()]
+        range_ids = _parallel_uploads(executor, uploader.process_substream_block, upload_tasks, running_futures)
+    else:
+        range_ids = [uploader.process_substream_block(b) for b in uploader.get_substream_blocks()]
+    return sorted(range_ids)
 
 
 class _ChunkUploader(object):  # pylint: disable=too-many-instance-attributes
@@ -229,6 +229,7 @@ def _upload_substream_block(self, block_id, block_stream):
 
     def _upload_substream_block_with_progress(self, block_id, block_stream):
         range_id = self._upload_substream_block(block_id, block_stream)
+        self._update_progress(len(block_stream))
         return range_id
 
     def set_response_properties(self, resp):
@@ -238,17 +239,23 @@ def set_response_properties(self, resp):
 
 class BlockBlobChunkUploader(_ChunkUploader):
 
+    def __init__(self, *args, **kwargs):
+        kwargs.pop('modified_access_conditions', None)
+        super(BlockBlobChunkUploader, self).__init__(*args, **kwargs)
+        self.current_length = None
+
     def _upload_chunk(self, chunk_offset, chunk_data):
         # TODO: This is incorrect, but works with recording.
-        block_id = encode_base64(url_quote(encode_base64('{0:032d}'.format(chunk_offset))))
+        index = '{0:032d}'.format(chunk_offset)
+        block_id = encode_base64(url_quote(encode_base64(index)))
         self.service.stage_block(
             block_id,
             len(chunk_data),
             chunk_data,
             data_stream_total=self.total_size,
             upload_stream_current=self.progress_total,
             **self.request_options)
-        return block_id
+        return index, block_id
 
     def _upload_substream_block(self, block_id, block_stream):
         try:

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

@@ -56,7 +56,7 @@ async def upload_data_chunks(
 
     if encryption_options:
         encryptor, padder = get_blob_encryptor_and_padder(
-            encryption_options.get('key'),
+            encryption_options.get('cek'),
             encryption_options.get('vector'),
             uploader_class is not PageBlobChunkUploader)
         kwargs['encryptor'] = encryptor

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

@@ -51,24 +51,13 @@ def _convert_mod_error(error):
     raise overwrite_error
 
 
-def _create_append_blob(
-        client,
-        blob_headers,
-        timeout,
-        access_conditions,
-        mod_conditions,
-        headers,
-        **kwargs):
-    created = client.create(
-        content_length=0,
-        blob_http_headers=blob_headers,
-        timeout=timeout,
-        lease_access_conditions=access_conditions,
-        modified_access_conditions=mod_conditions,
-        cls=return_response_headers,
-        headers=headers,
-        **kwargs)
-    ModifiedAccessConditions(if_match=created['etag'])  # TODO: Not working...
+def _any_conditions(modified_access_conditions=None, **kwargs):
+    return any([
+        modified_access_conditions.if_modified_since,
+        modified_access_conditions.if_unmodified_since,
+        modified_access_conditions.if_none_match,
+        modified_access_conditions.if_match
+    ])
 
 
 def upload_block_blob(  # pylint: disable=too-many-locals
@@ -84,8 +73,8 @@ def upload_block_blob(  # pylint: disable=too-many-locals
         encryption_options=None,
         **kwargs):
     try:
-        if not overwrite and not kwargs.get('modified_access_conditions'):
-            kwargs['modified_access_conditions'] = ModifiedAccessConditions(if_none_match='*')
+        if not overwrite and not _any_conditions(**kwargs):
+            kwargs['modified_access_conditions'].if_none_match = '*'
         adjusted_count = length
         if (encryption_options.get('key') is not None) and (adjusted_count is not None):
             adjusted_count += (16 - (length % 16))
@@ -123,7 +112,7 @@ def upload_block_blob(  # pylint: disable=too-many-locals
             if encryption_options.get('key'):
                 cek, iv, encryption_data = generate_blob_encryption_data(encryption_options['key'])
                 headers['x-ms-meta-encryptiondata'] = encryption_data
-                encryption_options['key'] = cek
+                encryption_options['cek'] = cek
                 encryption_options['vector'] = iv
             block_ids = upload_data_chunks(
                 service=client,
@@ -145,7 +134,6 @@ def upload_block_blob(  # pylint: disable=too-many-locals
                 max_connections=max_connections,
                 stream=stream,
                 validate_content=validate_content,
-                encryption_options=encryption_options,
                 **kwargs
             )
 
@@ -179,8 +167,8 @@ def upload_page_blob(
         encryption_options=None,
         **kwargs):
     try:
-        if not overwrite and not kwargs.get('modified_access_conditions'):
-            kwargs['modified_access_conditions'] = ModifiedAccessConditions(if_none_match='*')
+        if not overwrite and not _any_conditions(**kwargs):
+            kwargs['modified_access_conditions'].if_none_match = '*'
         if length is None or length < 0:
             raise ValueError("A content length must be specified for a Page Blob.")
         if length % 512 != 0:
@@ -227,49 +215,39 @@ def upload_page_blob(
 
 
 def upload_append_blob(
-        client,
-        stream,
-        length,
-        overwrite,
-        headers,
-        blob_headers,
-        access_conditions,
-        mod_conditions,
-        maxsize_condition,
-        validate_content,
-        timeout,
-        max_connections,
-        blob_settings,
-        **kwargs
-    ):
+        client=None,
+        stream=None,
+        length=None,
+        overwrite=None,
+        headers=None,
+        validate_content=None,
+        max_connections=None,
+        blob_settings=None,
+        encryption_options=None,
+        **kwargs):
     try:
         if length == 0:
             return {}
+        blob_headers = kwargs.pop('blob_headers', None)
         append_conditions = AppendPositionAccessConditions(
-            max_size=maxsize_condition,
+            max_size=kwargs.pop('maxsize_condition', None),
             append_position=None)
         try:
             if overwrite:
-                _create_append_blob(
-                    client,
-                    blob_headers,
-                    timeout,
-                    access_conditions,
-                    mod_conditions,
-                    headers,
+                client.create(
+                    content_length=0,
+                    blob_http_headers=blob_headers,
+                    headers=headers,
                     **kwargs)
             return upload_data_chunks(
-                blob_service=client,
-                blob_size=length,
-                block_size=blob_settings.max_block_size,
+                service=client,
+                uploader_class=AppendBlobChunkUploader,
+                total_size=length,
+                chunk_size=blob_settings.max_block_size,
                 stream=stream,
-                append_conditions=append_conditions,
                 max_connections=max_connections,
                 validate_content=validate_content,
-                access_conditions=access_conditions,
-                uploader_class=AppendBlobChunkUploader,
-                modified_access_conditions=mod_conditions,
-                timeout=timeout,
+                append_position_access_conditions=append_conditions,
                 **kwargs)
         except StorageErrorException as error:
             if error.response.status_code != 404:
@@ -282,26 +260,20 @@ def upload_append_blob(
                 except UnsupportedOperation:
                     # if body is not seekable, then retry would not work
                     raise error
-            _create_append_blob(
-                client,
-                blob_headers,
-                timeout,
-                access_conditions,
-                mod_conditions,
-                headers,
+            client.create(
+                content_length=0,
+                blob_http_headers=blob_headers,
+                headers=headers,
                 **kwargs)
             return upload_data_chunks(
-                blob_service=client,
-                blob_size=length,
-                block_size=blob_settings.max_block_size,
+                service=client,
+                uploader_class=AppendBlobChunkUploader,
+                total_size=length,
+                chunk_size=blob_settings.max_block_size,
                 stream=stream,
-                append_conditions=append_conditions,
                 max_connections=max_connections,
                 validate_content=validate_content,
-                access_conditions=access_conditions,
-                uploader_class=AppendBlobChunkUploader,
-                modified_access_conditions=mod_conditions,
-                timeout=timeout,
+                append_position_access_conditions=append_conditions,
                 **kwargs)
     except StorageErrorException as error:
         process_storage_error(error)

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

@@ -41,7 +41,7 @@
 from ._deserialize import deserialize_blob_properties, deserialize_blob_stream
 from ._upload_helpers import (
     upload_block_blob,
-    # upload_append_blob,
+    upload_append_blob,
     upload_page_blob)
 from .models import BlobType, BlobBlock
 from .lease import LeaseClient, get_access_conditions
@@ -495,24 +495,20 @@ def upload_blob(  # pylint: disable=too-many-locals
                 blob_settings=self._config,
                 encryption_options=encryption_options,
                 **kwargs)
-        # if blob_type == BlobType.AppendBlob:
-        #     if self.require_encryption or (self.key_encryption_key is not None):
-        #         raise ValueError(_ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION)
-        #     return upload_append_blob(
-        #         self._client.append_blob,
-        #         stream,
-        #         length,
-        #         overwrite,
-        #         headers,
-        #         blob_headers,
-        #         access_conditions,
-        #         mod_conditions,
-        #         maxsize_condition,
-        #         validate_content,
-        #         timeout,
-        #         max_connections,
-        #         self._config,
-        #         **kwargs)
+        if blob_type == BlobType.AppendBlob:
+            if self.require_encryption or (self.key_encryption_key is not None):
+                raise ValueError(_ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION)
+            return upload_append_blob(
+                client=self._client.append_blob,
+                stream=stream,
+                length=length,
+                overwrite=overwrite,
+                headers=headers,
+                validate_content=validate_content,
+                timeout=timeout,
+                max_connections=max_connections,
+                blob_settings=self._config,
+                **kwargs)
         raise ValueError("Unsupported BlobType: {}".format(blob_type))
 
     def download_blob(

sdk/storage/azure-storage-blob/tests/test_blob_samples_common.py

@@ -103,7 +103,7 @@ def test_soft_delete_and_undelete_blob(self):
         blob_service_client.set_service_properties(delete_retention_policy=delete_retention_policy)
 
         # Instantiate a ContainerClient
-        container_client = blob_service_client.get_container_client("containerforblobs")
+        container_client = blob_service_client.get_container_client("containerfordeletedblobs")
 
         # Create new Container
         try:
@@ -114,10 +114,7 @@ def test_soft_delete_and_undelete_blob(self):
 
         # Upload a blob to the container
         with open(SOURCE_FILE, "rb") as data:
-            container_client.upload_blob(name="my_blob", data=data)
-
-        # Get the blob client
-        blob_client = blob_service_client.get_blob_client("containerforblobs", "my_blob")
+            blob_client = container_client.upload_blob(name="my_blob", data=data)
 
         # Soft delete blob in the container (blob can be recovered with undelete)
         blob_client.delete_blob()
@@ -134,7 +131,7 @@ def test_soft_delete_and_undelete_blob(self):
         assert properties is not None
 
         # Delete container
-        blob_service_client.delete_container("containerforblobs")
+        blob_service_client.delete_container("containerfordeletedblobs")
 
     @record
     def test_acquire_lease_on_blob(self):

sdk/storage/azure-storage-blob/tests/test_common_blob.py

@@ -1474,7 +1474,7 @@ def test_download_to_file_with_credential(self):
         download_blob_from_url(
             source_blob.url, FILE_PATH,
             max_connections=2,
-            credential=self.settings.STORAGE_ACCOUNT_KEY)
+            credential=self.settings.REMOTE_STORAGE_ACCOUNT_KEY)
 
         # Assert
         with open(FILE_PATH, 'rb') as stream:
@@ -1495,7 +1495,7 @@ def test_download_to_stream_with_credential(self):
             download_blob_from_url(
                 source_blob.url, stream,
                 max_connections=2,
-                credential=self.settings.STORAGE_ACCOUNT_KEY)
+                credential=self.settings.REMOTE_STORAGE_ACCOUNT_KEY)
 
         # Assert
         with open(FILE_PATH, 'rb') as stream:
@@ -1514,7 +1514,7 @@ def test_download_to_file_with_existing_file(self):
         # Act
         download_blob_from_url(
             source_blob.url, FILE_PATH,
-            credential=self.settings.STORAGE_ACCOUNT_KEY)
+            credential=self.settings.REMOTE_STORAGE_ACCOUNT_KEY)
 
         with self.assertRaises(ValueError):
             download_blob_from_url(source_blob.url, FILE_PATH)
@@ -1536,13 +1536,13 @@ def test_download_to_file_with_existing_file_overwrite(self):
         # Act
         download_blob_from_url(
             source_blob.url, FILE_PATH,
-            credential=self.settings.STORAGE_ACCOUNT_KEY)
+            credential=self.settings.REMOTE_STORAGE_ACCOUNT_KEY)
 
         data2 = b'ABCDEFGH' * 1024 * 1024
         source_blob = self._create_remote_block_blob(blob_data=data2)
         download_blob_from_url(
             source_blob.url, FILE_PATH, overwrite=True,
-            credential=self.settings.STORAGE_ACCOUNT_KEY)
+            credential=self.settings.REMOTE_STORAGE_ACCOUNT_KEY)
 
         # Assert
         with open(FILE_PATH, 'rb') as stream: