diff --git a/cterasdk/direct/client.py b/cterasdk/direct/client.py index 4ac6282c..eb4862f5 100644 --- a/cterasdk/direct/client.py +++ b/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 diff --git a/cterasdk/direct/lib.py b/cterasdk/direct/lib.py index 3343f82c..778c23d8 100644 --- a/cterasdk/direct/lib.py +++ b/cterasdk/direct/lib.py @@ -83,33 +83,34 @@ 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. @@ -117,7 +118,10 @@ async def process_chunk(client, file_id, chunk, encryption_key, semaphore): :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,7 +145,7 @@ 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. @@ -149,6 +153,7 @@ async def process_chunks(client, file_id, chunks, encryption_key, semaphore=None :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 diff --git a/cterasdk/exceptions/direct.py b/cterasdk/exceptions/direct.py index 222be7ad..15097dcf 100644 --- a/cterasdk/exceptions/direct.py +++ b/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: diff --git a/docs/source/UserGuides/Miscellaneous/Changelog.rst b/docs/source/UserGuides/Miscellaneous/Changelog.rst index ed1f66c4..bab3795e 100644 --- a/docs/source/UserGuides/Miscellaneous/Changelog.rst +++ b/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 `_ + 2.20.45 -------