def chunk_list()

in azext_edge/edge/util/common.py [0:0]

• 15 lines of code
• 6 McCabe index (conditional complexity)

def chunk_list(data: list, chunk_len: int, data_size: int = 1024, size_unit: str = "kb") -> List[list]:
    if size_unit.lower() == "mb":
        data_size *= 1024

    result = []
    current_chunk = []

    for item in data:
        current_chunk.append(item)

        serialized_size = len(json.dumps(current_chunk).encode("utf-8")) / 1024  # convert bytes to kb

        if len(current_chunk) > chunk_len or serialized_size > data_size:
            current_chunk.pop()
            result.append(current_chunk)
            current_chunk = [item]

    if current_chunk:
        result.append(current_chunk)

    return result