KV Store Serialize/Deserialize

Serialization Function

The serialize function converts a dictionary into a length-prefixed string format. For each key-value pair, encode as keyLen:keyData:valueLen:valueData. For example, {"ab": "xyz"} becomes 2:ab3:xyz. Iterate through the dictionary, calculate string lengths using len(), and concatenate the formatted segments. Handle empty dictionaries by returning an empty string.

def serialize(kv_dict):
    if not kv_dict:
        return ""
    
    result = []
    for key, value in kv_dict.items():
        key_str = str(key)
        value_str = str(value)
        segment = f"{len(key_str)}:{key_str}{len(value_str)}:{value_str}"
        result.append(segment)
    
    return "".join(result)

Deserialization Function

The deserialize function parses the encoded string back into a dictionary. Use a position pointer to track the current parsing location. Read characters until : to extract the length integer, then extract exactly that many characters as data. Repeat this process twice per key-value pair (once for key, once for value). For example, parsing 2:ab3:xyz reads 2, extracts ab, reads 3, extracts xyz, yielding {"ab": "xyz"}.

def deserialize(s):
    kv = {}
    pos = 0
    while pos < len(s):
        colon = s.index(':', pos)
        key_len = int(s[pos:colon])
        pos = colon + 1
        key = s[pos:pos + key_len]
        pos += key_len
        
        colon = s.index(':', pos)
        val_len = int(s[pos:colon])
        pos = colon + 1
        val = s[pos:pos + val_len]
        pos += val_len
        
        kv[key] = val
    return kv

Length Parsing Helper

Create a helper function read_length_and_data(s, pos) that extracts one length-prefixed segment. Starting at position pos, scan forward to find :, parse the integer length, then extract the substring of that exact length. Return both the extracted data and the new position. This encapsulates the core parsing logic and makes deserialization cleaner by handling one segment at a time.

def read_length_and_data(s, pos):
    colon_pos = s.index(':', pos)
    length = int(s[pos:colon_pos])
    data_start = colon_pos + 1
    data_end = data_start + length
    data = s[data_start:data_end]
    return data, data_end

Chunked Serialization with Metadata Tracking

The core challenge shifts from serializing the entire KV store as one blob to splitting it into fixed-size chunks (e.g., 1KB each) for upload or storage. This matters because real systems often have size constraints on network packets, API payloads, or storage blocks. You'll need a way to track how many chunks exist so deserialization knows when to stop reading.

The key insight is introducing a metadata file that stores critical information like total_chunks count before writing the actual data chunks. During serialization, you write the metadata first, then split your serialized KV data into 1KB chunks as separate files or blocks. During deserialization, you read the metadata to know exactly how many chunks to fetch and reassemble.

Important considerations include:

• Handling the last chunk which will likely be smaller than 1KB
• Deciding whether to split on byte boundaries (simple) or respect character boundaries to avoid corrupting multi-byte UTF-8 sequences
• Choosing a naming scheme for chunk files (e.g., data_chunk_0, data_chunk_1) that preserves order
• Ensuring atomic writes of metadata so you never have orphaned chunks

import json
import os
CHUNK_SIZE = 1024
def serialize_chunked(kv_store, base_path):
    serialized = json.dumps(kv_store)
    data_bytes = serialized.encode('utf-8')
    total_chunks = (len(data_bytes) + CHUNK_SIZE - 1) // CHUNK_SIZE
    
    metadata = {'total_chunks': total_chunks}
    with open(f'{base_path}_metadata.json', 'w') as f:
        json.dump(metadata, f)
    
    for i in range(total_chunks):
        chunk = data_bytes[i * CHUNK_SIZE:(i + 1) * CHUNK_SIZE]
        with open(f'{base_path}_chunk_{i}', 'wb') as f:
            f.write(chunk)
def deserialize_chunked(base_path):
    with open(f'{base_path}_metadata.json', 'r') as f:
        metadata = json.load(f)
    
    data_bytes = b''
    for i in range(metadata['total_chunks']):
        with open(f'{base_path}_chunk_{i}', 'rb') as f:
            data_bytes += f.read()
    
    return json.loads(data_bytes.decode('utf-8'))