PyBite
Chunk by chunk iteration made easier
Installation
pip install pybite
Methods
iterate_by
Return a iterator of chunks for the iterable
Parameters
iterable: iter
Any iterable data e.g. list, tuple, dict, iter, ...
chunk_size: int
The size of each chunk
map: callable optional, defaults None
A map function for transform the data before dividing in chunks
persist_header: bool optional, defaults False
Whether to persist a header on each chunk or not. If persist_header is True and header is None, the firt line will be read as the actual header.
header: str optional, defaults None
A header to be written at the start of each file. If persist_header is True and header is None, the firt line will be read as the actual header.
Returns
iter
New iterable for the chunked data
Examples
>>> iterate_by([1, 2, 3, 4, 5], 2)
iter([[1, 2], [3, 4], [5]])
>>> iterate_by([1, 2, 3, 4, 5], 2, map=lambda x: x * 2)
iter([[2, 4], [6, 8], [10]])
>>> iterate_by(["numbers", 1, 2, 3, 4, 5], 2, persist_header=True)
iter([["numbers", 1, 2], ["numbers", 3, 4], ["numbers", 5]])iterate_file_by_lines
Return a iterator of file lines.
Each line is read on-demand untill there is no more line to be read.
Parameters
iterable: iter
Any iterable data e.g. list, tuple, dict, iter, ...
file_stream : str, io.StringIO
A file path or io.StringIO instance to the file to be read.
encoding : str optional, defaults None
The input file encoding. The chunks will be saved using the same encoding.
strip_end : bool optional, defaults False
Flag for strip the end of each line.
Same as calling line.rstrip().
Returns
iter
New iterable for the chunked data
Examples
with open("test.txt", "w", encoding="utf-8") as f:
f.write("Symbols\nAyp\nBx\nCC\nDt")
>>> iterate_file_by_lines("test.txt", encoding="utf-8")
iter(["Symbols\n", "Ayp\n", "Bx\n", "CC\n", "Dt"])
>>> iterate_file_by_lines("test.txt", encoding="utf-8", strip_end=True)
iter(["Symbols", "Ayp", "Bx", "CC", "Dt"])split_file_by_lines
Split a file into multiple files and store them in the output path.
The file is divided b lines into chunk files that later can be read individually or joined back.
Parameters
file_stream: str, io.StringIO
A file path or io.StringIO instance to the file you want to split.
output_path: str
The path to the directory where the chunks will be stored.
lines: int
The quantity of lines on each chunk file. If included the header the total will be lines + 1.
encoding: str optional, defaults None
The input file encoding. The chunks will be saved using the same encoding.
persist_header: bool optional, defaults False
Whether to persist a header on each chunk or not. If persist_header is True and header is None, the firt line will be read as the actual header.
header: str optional, defaults None
A header to be written at the start of each file. If persist_header is True and header is None, the firt line will be read as the actual header.
chunk_name_format: str optional, defaults "04d"
The format string for the chunk numbers in the output file names.
Returns
list
List of paths to the written files.
Examples
with open("test.txt", "w", encoding="utf-8") as f:
f.write("Symbols\nAyp\nBx\nCC\nDt")
>>> split_by_lines("test.txt", "out", 2, encoding="utf-8")
["out/test.chunk0000.txt", "out/test.chunk0001.txt",
"out/test.chunk0002.txt"]
>>> split_by_lines("test.txt", "out", 2, encoding="utf-8", persist_header=True)
["out/test.chunk0000.txt", "out/test.chunk0001.txt"]join_file_chunks
Join chunk files into a single line stream.
Join the files created by split_by_lines into a iterable of str lines and read ordered by name. Chunks not found will throw an error if ignore_missing_chunks is not False.
Avoid saving different files chunks into the same directory.
Parameters
files_path: str
The path to a directory containing the chunk files or a list of the path to the files.
encoding: str optional, defaults None
The input file encoding. The chunks will be saved using the same encoding.
persisted_header: bool optional, defaults False
Whether a header was persisted on each chunk or not. If persisted_header is True and header is None, the firt line of teh first file will be read as the actual header.
header: str optional, defaults None
A header to be read at the start of the first file. If persisted_header is True and header is None, the firt line of teh first file will be read as the actual header.
ignore_missing_chunks: bool optional, defaults "False
Flag to ignore missing chunks.
Returns
iter
An iterable to the lines of the files (read in the order).
split_file_by_lines
Slice file by lines into an iterable of the sliced lines.
Returns only cut lines in iterable str format. It does not work with negative numbers.
Parameters
files_stream: str
The path and file to be cut.
start: int
Initial position of the cut. Use positive Number.
end: int
End position of the cut. Use Positive Number.
encoding: str optional, defaults None
The input file encoding. The chunks will be saved using the same encoding.
persisted_header: bool optional, defaults False
Whether a header was persisted on each chunk or not. If persisted_header is True and header is None, the firt line of teh first file will be read as the actual header.
header: str optional, defaults None
A header to be read at the start of the first file. If persisted_header is True and header is None, the firt line of teh first file will be read as the actual header.
Returns
iter
An iterable with the cut portion of the file
Examples
test_data = [
"Name;Age;",
"Test;00;",
"Test;11;",
"John;22;",
"Test;33;",
"Test;44;",
"Test;55;",
"Test;66;",
"Test;77;",
"Test;88;",
]
with open("./text.txt", "w") as f:
f.write("\n".join(test_data))
>>> slice_file_by_lines(input_path, 1,2, persist_header=True)
iter(["Name;Age;\n", "Test;11;\n", "John;22;\n"])
>>> slice_file_by_lines(input_path, 1,2)
iter(["Test;11;\n", "John;22;\n"])Test
Test are handled by PyTest and are included inside the folder test.
To run the test execute the command:
pytest
