How to Mock Kafka Topic Partitions with a Python dict of lists

from collections import defaultdict

class KafkaTopicPartitionMock:
    """A simple mock for Kafka topic-partition assignment using dict of lists."""

    def __init__(self, topic):
        self.topic = topic
        self.partitions = defaultdict(list)  # partition_id -> list of messages

    def produce(self, message, partition=None):
        """Assign message to a partition. If None, use round-robin."""
        if partition is None:
            partition = len(self.partitions) % 3  # mock 3 partitions
        self.partitions[partition].append(message)

    def consume(self, partition):
        """Retrieve all messages from a partition."""
        return self.partitions.get(partition, [])

    def show_partition_counts(self):
        """Display message counts per partition."""
        return {p: len(msgs) for p, msgs in sorted(self.partitions.items())}


if __name__ == "__main__":
    topic = KafkaTopicPartitionMock("orders")
    topic.produce("order-1", partition=0)
    topic.produce("order-2", partition=1)
    topic.produce("order-3", partition=2)
    topic.produce("order-4", partition=0)
    topic.produce("order-5", partition=1)

    print("Partition counts:", topic.show_partition_counts())
    print("Partition 0:", topic.consume(0))
    print("Partition 1:", topic.consume(1))
    print("Partition 2:", topic.consume(2))