Compression

QuantConnect.Compression

Bases: Object

Compression class manages the opening and extraction of compressed files (zip, tar, tar.gz).

get_zip_entry_file_names

get_zip_entry_file_names(
    zip_file_name: str,
) -> Iterable[str]

get_zip_entry_file_names(
    zip_file_stream: Stream,
) -> Iterable[str]

Signature descriptions:

• Returns the entry file names contained in a zip file

• Return the entry file names contained in a zip file

Parameters:

Name	Type	Description	Default
zip_file_name	Optional[str]	The zip file name	None
zip_file_stream	Optional[Stream]	Stream to the file	None

Returns:

Type	Description
Iterable[str]	Depends on the signature used. Case 1: [An IEnumerable of entry file names.]; Case 2: [IEnumerable of entry file names.]

un_tar

un_tar(
    stream: Stream, is_tar_gz: bool
) -> Iterable[KeyValuePair[str, List[int]]]

un_tar(
    source: str,
) -> Iterable[KeyValuePair[str, List[int]]]

Signature descriptions:

• Enumerate through the files of a TAR and get a list of KVP names-byte arrays

• Enumerate through the files of a TAR and get a list of KVP names-byte arrays.

Parameters:

Name	Type	Description	Default
stream	Optional[Stream]	The input tar stream	None
is_tar_gz	Optional[bool]	True if the input stream is a .tar.gz or .tgz	None
source	Optional[str]		None

Returns:

Type	Description
Iterable[KeyValuePair[str, List[int]]]	An enumerable containing each tar entry and it's contents.

unzip

unzip(
    zip: str, directory: str, overwrite: bool = False
) -> bool

unzip(
    filename: str, zip: Optional[Any]
) -> Tuple[StreamReader, Any]

unzip(
    filename: str, zip_entry_name: str, zip: Optional[Any]
) -> Tuple[StreamReader, Any]

unzip(
    filename: str,
) -> Iterable[KeyValuePair[str, List[str]]]

unzip(
    stream: Stream,
) -> Iterable[KeyValuePair[str, List[str]]]

Signature descriptions:

• Unzips the specified zip file to the specified directory

• Streams a local zip file using a streamreader. Important: the caller must call Dispose() on the returned ZipFile instance.

• Streams the unzipped file as key value pairs of file name to file contents. NOTE: When the returned enumerable finishes enumerating, the zip stream will be closed rendering all key value pair Value properties unaccessible. Ideally this would be enumerated depth first.

• Lazily unzips the specified stream

Parameters:

Name	Type	Description	Default
zip	Optional[str | Optional[Any]]	The zip to be unzipped	None
directory	Optional[str]	The directory to place the unzipped files	None
overwrite	Optional[bool]	Flag specifying whether or not to overwrite existing files	False
filename	Optional[str]	Location of the original zip file	None
zip_entry_name	Optional[str]	The zip entry name to open a reader for. Specify null to access the first entry	None
stream	Optional[Stream]	The zipped stream to be read	None

Returns:

Type	Description
bool | Tuple[StreamReader, Any] | Iterable[KeyValuePair[str, List[str]]]	Depends on the signature used. Case 1: [Stream reader of the first file contents in the zip file.]; Case 2: [The stream zip contents.]; Case 3: [An enumerable whose elements are zip entry key value pairs with]

unzip_to_folder

unzip_to_folder(
    zip_data: List[int], output_folder: str
) -> List[str]

unzip_to_folder(zip_file: str) -> List[str]

Signature descriptions:

• Unzip the given byte array and return the created file names.

• Unzip a local file and return the created file names

Parameters:

Name	Type	Description	Default
zip_data	Optional[List[int]]	A byte array containing the zip	None
output_folder	Optional[str]	The target output folder	None
zip_file	Optional[str]	Location of the zip on the HD	None

Returns:

Type	Description
List[str]	List of unzipped file names.

zip

zip(
    text_path: str,
    zip_entry_name: str,
    delete_original: bool = True,
) -> str

zip(
    source: str,
    destination: str,
    zip_entry_name: str,
    delete_original: bool,
) -> None

zip(text_path: str, delete_original: bool = True) -> str

zip(data: str, zip_path: str, zip_entry: str) -> None

Signature descriptions:

• Compress a given file and delete the original file. Automatically rename the file to name.zip.

• Compresses the specified source file.

• Compress given data to the path given

Parameters:

Name	Type	Description	Default
text_path	Optional[str]	Path of the original file	None
zip_entry_name	Optional[str]	The name of the entry inside the zip file	None
delete_original	Optional[bool]	Boolean flag to delete the original file after completion	True
source	Optional[str]	The source file to be compressed	None
destination	Optional[str]	The destination zip file path	None
data	Optional[str]	Data to write to zip	None
zip_path	Optional[str]	Path to write to	None
zip_entry	Optional[str]	Entry to save the data as	None

Returns:

Type	Description
Optional[str]	String path for the new zip file.

zip_bytes_async

zip_bytes_async(
    target: Stream,
    data: List[int],
    zip_entry_name: str,
    mode: Optional[ZipArchiveMode] = None,
    compression_level: Optional[CompressionLevel] = None,
) -> Task

zip_bytes_async(
    target: Stream,
    data: List[KeyValuePair[List[int], str]],
    mode: Optional[ZipArchiveMode] = None,
    compression_level: Optional[CompressionLevel] = None,
) -> Task

Performs an in memory zip of the specified bytes in the target stream

Parameters:

Name	Type	Description	Default
target	Stream	The target stream	required
data	List[int] | List[KeyValuePair[List[int], str]]	The file contents in bytes to be zipped	required
zip_entry_name	Optional[str]	The zip entry name	None
mode	Optional[ZipArchiveMode]	The archive mode	None
compression_level	Optional[CompressionLevel]	The desired compression level	None

Returns:

Type	Description
Task	The zipped file as a byte array.

zip_create_append_data

zip_create_append_data(
    path: str,
    entry: str,
    data: str,
    override_entry: bool = False,
) -> bool

zip_create_append_data(
    path: str,
    entry: str,
    data: List[int],
    override_entry: bool = False,
) -> bool

Append the zip data to the file-entry specified.

Parameters:

Name	Type	Description	Default
path	str	The zip file path	required
entry	str	The entry name	required
data	str | List[int]	The entry data	required
override_entry	bool	True if should override entry if it already exists	False

Returns:

Type	Description
bool	True on success.

zip_data

zip_data(
    zip_path: str, filenames_and_data: Dictionary[str, str]
) -> bool

zip_data(
    zip_path: str,
    filenames_and_data: List[KeyValuePair[str, List[int]]],
) -> bool

zip_data(
    zip_path: str, zip_entry: str, lines: List[str]
) -> bool

Signature descriptions:

• Create a zip file of the supplied file names and string data source

• Create a zip file of the supplied file names and data using a byte array

• Zips the specified lines of text into the zip_path

Parameters:

Name	Type	Description	Default
zip_path	str	Output location to save the file.	required
filenames_and_data	Optional[Dictionary[str, str] | List[KeyValuePair[str, List[int]]]]	File names and data in a dictionary format.	None
zip_entry	Optional[str]	The entry name in the zip	None
lines	Optional[List[str]]	The lines to be written to the zip	None

Returns:

Type	Description
bool	Depends on the signature used. Case 1: [True on successfully creating the zip file.]; Case 2: [True on successfully saving the file.]; Case 3: [True if successful, otherwise false.]

zip_streams_async

zip_streams_async(
    target: str,
    data: List[KeyValuePair[str, Stream]],
    mode: Optional[ZipArchiveMode] = None,
    compression_level: Optional[CompressionLevel] = None,
) -> Task

zip_streams_async(
    target: Stream,
    data: List[KeyValuePair[str, Stream]],
    mode: Optional[ZipArchiveMode] = None,
    compression_level: Optional[CompressionLevel] = None,
    leave_stream_open: bool = False,
) -> Task

Performs an in memory zip of the specified stream in the target stream

Parameters:

Name	Type	Description	Default
target	str | Stream	The target stream	required
data	List[KeyValuePair[str, Stream]]	The file contents in bytes to be zipped	required
mode	Optional[ZipArchiveMode]	The archive mode	None
compression_level	Optional[CompressionLevel]	The desired compression level	None
leave_stream_open	Optional[bool]	True to leave the taget stream open	False

Returns:

Type	Description
Task	The zipped file as a byte array.

extract_7_zip_archive

extract_7_zip_archive(
    input_file: str,
    output_directory: str,
    exec_timeout: int = 60000,
) -> None

Extracts a 7-zip archive to disk, using the 7-zip CLI utility

Parameters:

Name	Type	Description	Default
input_file	str	Path to the 7z file	required
output_directory	str	Directory to output contents of 7z	required
exec_timeout	int	Timeout in seconds for how long we should wait for the extraction to complete	60000

read_lines

read_lines(filename: str) -> List[str]

Streams each line from the first zip entry in the specified zip file

Parameters:

Name	Type	Description	Default
filename	str	The zip file path to stream	required

Returns:

Type	Description
List[str]	An enumerable containing each line from the first unzipped entry.

un_g_zip

un_g_zip(gzip_file_name: str, target_directory: str) -> str

Extract .gz files to disk

Parameters:

Name	Type	Description	Default
gzip_file_name	str		required
target_directory	str		required

un_tar_files

un_tar_files(source: str, destination: str) -> None

Extracts all file from a zip archive and copies them to a destination folder.

Parameters:

Name	Type	Description	Default
source	str	The source zip file.	required
destination	str	The destination folder to extract the file to.	required

un_tar_gz_files

un_tar_gz_files(source: str, destination: str) -> None

Extract tar.gz files to disk

Parameters:

Name	Type	Description	Default
source	str	Tar.gz source file	required
destination	str	Location folder to unzip to	required

unzip_data

unzip_data(
    zip_data: List[int], encoding: Encoding = None
) -> Dictionary[str, str]

Uncompress zip data byte array into a dictionary string array of filename-contents.

Parameters:

Name	Type	Description	Default
zip_data	List[int]	Byte data array of zip compressed information	required
encoding	Encoding	Specifies the encoding used to read the bytes. If not specified, defaults to ASCII	None

Returns:

Type	Description
Dictionary[str, str]	Uncompressed dictionary string-sting of files in the zip.

unzip_data_async

unzip_data_async(
    stream: Stream, encoding: Encoding = None
) -> Task[Dictionary[str, str]]

Uncompress zip data byte array into a dictionary string array of filename-contents.

Parameters:

Name	Type	Description	Default
stream	Stream	Stream data of zip compressed information	required
encoding	Encoding	Specifies the encoding used to read the bytes. If not specified, defaults to ASCII	None

Returns:

Type	Description
Task[Dictionary[str, str]]	Uncompressed dictionary string-sting of files in the zip.

unzip_stream

unzip_stream(
    zipstream: Stream,
    zip_file: Optional[Any],
    entry_name: str = None,
) -> Tuple[Stream, Any]

Unzip a stream that represents a zip file and return the first entry as a stream

unzip_stream_to_stream_reader

unzip_stream_to_stream_reader(
    zipstream: Stream,
) -> StreamReader

Unzip a local file and return its contents via streamreader:

validate_zip

validate_zip(path: str) -> bool

Validates whether the zip is corrupted or not

Parameters:

Name	Type	Description	Default
path	str	Path to the zip file	required

Returns:

Type	Description
bool	true if archive tests ok; false otherwise.

zip_bytes

zip_bytes(
    bytes: List[int], zip_entry_name: str
) -> List[int]

Performs an in memory zip of the specified bytes

Parameters:

Name	Type	Description	Default
bytes	List[int]	The file contents in bytes to be zipped	required
zip_entry_name	str	The zip entry name	required

Returns:

Type	Description
List[int]	The zipped file as a byte array.

zip_directory

zip_directory(
    directory: str,
    destination: str,
    include_root_in_zip: bool = True,
) -> bool

Zips the specified directory, preserving folder structure

Parameters:

Name	Type	Description	Default
directory	str	The directory to be zipped	required
destination	str	The output zip file destination	required
include_root_in_zip	bool	True to include the root 'directory' in the zip, false otherwise	True

Returns:

Type	Description
bool	True on a successful zip, false otherwise.

zip_files

zip_files(destination: str, files: List[str]) -> None

Zips all files specified to a new zip at the destination path