import struct
[docs]
class SnowflakeId:
"""
see https://en.wikipedia.org/wiki/Snowflake_ID
"""
SNOWFLAKE_ID_WIDTH = (
10 # with in bytes (8 bits each), can be more than long int width (8 bytes)
)
[docs]
def __init__(self, hi_bits_or_bytes=None, lo_bits=None):
self.bits = [0, 0]
if lo_bits is None:
if hi_bits_or_bytes is not None:
i = 0
shift = 8 # starting from Hi value, which is only 2 bytes wide
n = 1
while i < len(hi_bits_or_bytes) and i <= self.SNOWFLAKE_ID_WIDTH:
self.bits[n] |= hi_bits_or_bytes[i] << shift
if shift == 0:
n -= 1
shift = 56
else:
shift -= 8
i += 1
else:
self.bits = [lo_bits, hi_bits_or_bytes]
[docs]
@staticmethod
def from_hex(_input):
expected_length = 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH
if _input is None or len(_input) < expected_length:
raise RuntimeError(
f"MessageId is {SnowflakeId.SNOWFLAKE_ID_WIDTH * 8} bits wide, it is represented by HEX string "
f"of length: {expected_length}, real input: [{_input}] of length: {len(_input) if _input else 0}"
)
bits = [0, 0]
n = 1
for i in range(0, expected_length, 2):
if i == expected_length - (2 * 8):
n = 0
bits[n] = (bits[n] << 8) | int(_input[i : i + 2], 16)
return SnowflakeId(bits[1], bits[0])
def __str__(self):
_hex = f"{self.bits[1]:016X}{self.bits[0]:016X}"
_val = (
_hex
if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH
else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH) :]
)
return _val.lower()
def __eq__(self, other):
if isinstance(other, SnowflakeId):
return self.bits[1] == other.bits[1] and self.bits[0] == other.bits[0]
return False
def __lt__(self, other):
if self.bits[1] < other.bits[1]:
return True
elif self.bits[1] > other.bits[1]:
return False
return self.bits[0] < other.bits[0]
[docs]
def get_bytes(self):
high = struct.pack("!Q", self.bits[1])
low = struct.pack("!Q", self.bits[0])
return high[-2:] + low
[docs]
def get_bits(self):
return self.bits
[docs]
def as_string(self) -> str:
return self.__str__()