def process_arrow_stream_message()

in http/get_multipart/python/client/simple_client.py [0:0]

• 20 lines of code
• 3 McCabe index (conditional complexity)

def process_arrow_stream_message(message):
    global arrow_stream_parsing_time
    assert message.get_content_type() == ARROW_STREAM_FORMAT
    payload = part.get_payload(decode=True)
    print(f"-- {len(payload)} bytes of Arrow data:")
    num_batches = 0
    num_records = 0
    start_time = time.time()
    with pa.ipc.open_stream(payload) as reader:
        schema = reader.schema
        print(f"Schema: \n{schema}\n")
        try:
            while True:
                batch = reader.read_next_batch()
                num_batches += 1
                num_records += batch.num_rows
        except StopIteration:
            pass
    arrow_stream_parsing_time = time.time() - start_time
    print(f"Parsed {num_records} records in {num_batches} batch(es)")