SDK-373 Skip Direct IO decrypt and decompress when the portal reports a plaintext native object

cterasdk/direct/client.py

@@ -73,6 +73,7 @@ async def streamer(self, file_id, byte_range=None, max_workers=None):
         :returns: Streamer object.
         :rtype: cterasdk.direct.stream.Streamer
         """
+        byte_range = byte_range if byte_range is not None else ByteRange.default()
         executor = self.executor(await self._chunks(file_id), byte_range=byte_range, max_workers=max_workers)
         return Streamer(executor, byte_range)
 
@@ -95,7 +96,9 @@ async def execute():
             """
             Asynchronous Executable of Chunk Retrieval Tasks.
             """
-            return await process_chunks(self._client, file_id, chunks, metadata.encryption_key,
+            return await process_chunks(self._client, file_id, chunks,
+                                        metadata.encryption_key,
+                                        metadata.compression_library,
                                         asyncio.Semaphore(max_workers) if max_workers else None)
 
         return execute

cterasdk/direct/lib.py

@@ -83,41 +83,45 @@ async def decrypt_object(file_id, encrypted_object, encryption_key, chunk):
     :rtype: bytes
     """
     try:
-        return decrypt_block(encrypted_object, encryption_key)
+        return decrypt_block(encrypted_object, encryption_key) if encryption_key is not None else encrypted_object
     except DirectIOError:
         logger.error('Failed to decrypt block.')
         raise DecryptBlockError(file_id, chunk)
 
 
-async def decompress_object(file_id, compressed_object, chunk):
+async def decompress_object(file_id, compressed_object, compression_library, chunk):
     """
     Decompress Object.
 
     :param bytes compressed_object: Compressed object.
+    :param str compression_library: Compression library.
     :param cterasdk.direct.types.Chunk chunk: Chunk.
     :returns: Decompressed Object.
     :rtype: bytes
     """
     try:
-        decompressed_object = decompress(compressed_object)
+        decompressed_object = decompress(compressed_object) if compression_library is not None else compressed_object
         if chunk.length != len(decompressed_object):
             logger.error('Expected block length does not match decrypted and decompressed block length.')
             raise BlockValidationException(file_id, chunk)
         return decompressed_object
-    except DirectIOError:
+    except DirectIOError as e:
         logger.error('Failed to decompress block.')
-        raise DecompressBlockError(file_id, chunk)
+        raise DecompressBlockError(file_id, chunk) from e
 
 
-async def process_chunk(client, file_id, chunk, encryption_key, semaphore):
+async def process_chunk(client, file_id, chunk, encryption_key, compression_library, semaphore):
     """
     Process a Chunk.
 
     :param cterasdk.clients.clients.AsyncClient client: Asynchronous HTTP Client.
     :param int file_id: File ID.
     :param cterasdk.direct.types.Chunk chunk: Chunk.
     :param str encryption_key: Encryption key.
+    :param str compression_library: Compression library.
     :param asyncio.Semaphore semaphore: Semaphore.
+    :param bool,optional encrypted: Decrypt the downloaded object. Defaults to ``True``.
+    :param bool,optional compressed: Decompress the downloaded object. Defaults to ``True``.
 
     :returns: Block
     :rtype: cterasdk.direct.types.Block
@@ -132,7 +136,7 @@ async def process(client, chunk, encryption_key):
         logger.debug(message)
         encrypted_object = await get_object(client, file_id, chunk)
         decrypted_object = await decrypt_object(file_id, encrypted_object, encryption_key, chunk)
-        decompressed_object = await decompress_object(file_id, decrypted_object, chunk)
+        decompressed_object = await decompress_object(file_id, decrypted_object, compression_library, chunk)
         return Block(file_id, chunk.number, chunk.offset, decompressed_object, chunk.length)
 
     if semaphore is not None:
@@ -141,14 +145,15 @@ async def process(client, chunk, encryption_key):
     return await process(client, chunk, encryption_key)
 
 
-async def process_chunks(client, file_id, chunks, encryption_key, semaphore=None):
+async def process_chunks(client, file_id, chunks, encryption_key, compression_library, semaphore=None):
     """
     Process Chunks Asynchronously.
 
     :param cterasdk.clients.clients.AsyncClient client: Asynchronous HTTP Client.
     :param int file_id: File ID.
     :param list[cterasdk.direct.types.Chunk] chunks: Chunk.
     :param str encryption_key: Encryption key.
+    :param str compression_library: Compression library.
     :param asyncio.Semaphore,optional semaphore: Semaphore.
     :returns: List of futures.
     :rtype: list[asyncio.Task]
@@ -161,7 +166,8 @@ async def process_chunks(client, file_id, chunks, encryption_key, semaphore=None
     logger.debug(' '.join(message))
     futures = []
     for chunk in chunks:
-        futures.append(asyncio.create_task(process_chunk(client, file_id, chunk, encryption_key, semaphore)))
+        futures.append(asyncio.create_task(process_chunk(
+            client, file_id, chunk, encryption_key, compression_library, semaphore)))
     return futures
 
 

cterasdk/exceptions/direct.py

@@ -116,7 +116,7 @@ def __init__(self, file_id, chunk):
 class BlockValidationException(BlockError):
 
     def __init__(self, file_id, chunk):
-        super().__init__(errno.EIO, 'Expected block length does not match decrypted and decompressed block length', file_id, chunk)
+        super().__init__(errno.EIO, 'Expected block length does not match downloaded block length', file_id, chunk)
 
 
 class BlockInfo:

docs/source/UserGuides/Miscellaneous/Changelog.rst

@@ -1,6 +1,16 @@
 Changelog
 =========
 
+2.20.46
+-------
+
+Improvements
+^^^^^^^^^^^^
+
+* Support CTERA Direct I/O download for files stored in Fusion Direct cloudfolders
+
+Related issues and pull requests on GitHub: `#369 <https://github.com/ctera/ctera-python-sdk/pull/369>`_
+
 2.20.45
 -------