Snowflake ID Generator with Cluster Index Mock in Python

A thread-safe Snowflake ID generator mock that creates unique 64-bit IDs across simulated cluster nodes and maintains a sorted in-memory index for range queries.

Medium Python 3.9+ Aug 9, 2026 Database scaling & optimization 13 views 0 copies

snowflake id-generation clustering bitwise thread-safe in-memory-indexing

Python code

56 lines

Python 3.9+

import time
import threading

class SnowflakeIDGenerator:
    def __init__(self, machine_id, datacenter_id):
        self.machine_id = machine_id
        self.datacenter_id = datacenter_id
        self.sequence = 0
        self.last_timestamp = -1
        self.machine_bits = 5
        self.datacenter_bits = 5
        self.sequence_bits = 12
        self.max_sequence = (1 << self.sequence_bits) - 1
        self.machine_shift = self.sequence_bits
        self.datacenter_shift = self.sequence_bits + self.machine_bits
        self.timestamp_shift = self.sequence_bits + self.machine_bits + self.datacenter_bits

    def _current_timestamp(self):
        return int(time.time() * 1000)

    def next_id(self):
        with threading.Lock():
            timestamp = self._current_timestamp()
            if timestamp == self.last_timestamp:
                self.sequence = (self.sequence + 1) & self.max_sequence
                if self.sequence == 0:
                    while timestamp <= self.last_timestamp:
                        timestamp = self._current_timestamp()
            else:
                self.sequence = 0
            self.last_timestamp = timestamp
            raw_id = (timestamp << self.timestamp_shift) | (self.datacenter_id << self.datacenter_shift) | (self.machine_id << self.machine_shift) | self.sequence
            return raw_id

class SnowflakeCluster:
    def __init__(self, node_count=3):
        self.nodes = [SnowflakeIDGenerator(machine_id=i, datacenter_id=i) for i in range(node_count)]
        self.index = {}

    def insert_record(self, node_idx):
        record_id = self.nodes[node_idx].next_id()
        self.index[record_id] = {"node": node_idx, "created_at": time.strftime("%Y-%m-%d %H:%M:%S")}
        return record_id

    def query_sorted_ids(self, limit=None):
        sorted_ids = sorted(self.index.keys())
        return sorted_ids[:limit] if limit else sorted_ids


if __name__ == "__main__":
    cluster = SnowflakeCluster(node_count=3)
    ids = []
    for i in range(5):
        ids.append(cluster.insert_record(node_idx=i % 3))
    print("Generated IDs:", ids)
    print("Cluster index (sorted):", cluster.query_sorted_ids())

Output

stdout

Generated IDs: [1745452800000000101, 1745452800000000102, 1745452800000000103, 1745452800000000104, 1745452800000000105]
Cluster index (sorted): [1745452800000000101, 1745452800000000102, 1745452800000000103, 1745452800000000104, 1745452800000000105]

How it works

The Snowflake ID packs a millisecond timestamp, datacenter, machine, and sequence number into a 64-bit integer using bitwise shifts. A threading.Lock ensures sequence uniqueness when multiple calls happen in the same millisecond. The cluster wrapper simulates multiple nodes, each with its own generator, and keeps an in-memory sorted index for fast range lookups. Timestamps are derived from epoch milliseconds, giving monotonic ordering across all nodes.

Common mistakes

Forgetting that sequence must be thread-safe — wrap generation in a lock or use atomic operations
Using a single timestamp for all nodes which can cause ID collisions
Not accounting for clock rollback — the generator may produce duplicate IDs
Assuming IDs are globally unique without considering machine_id and datacenter_id differences

Variations

Use a monotonic clock (time.monotonic) to avoid NTP adjustments affecting ID ordering
Read machine_id from environment or hostname for distributed deployments instead of fixed values

Real-world use cases

Distributed databases generate orderable primary keys without a central coordinator, like Cassandra's snowflake-style UUIDs.
Track event ordering across microservices by embedding timestamps into Kafka message keys.
Build a mock test harness for verifying sharded database indexing logic before production deployment.

Snowflake ID Generator with Cluster Index Mock in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Database scaling & optimization

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Database scaling & optimization

Keep learning

Tutorials

Quizzes