Stanislav Hejny
73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
import struct
|
|
|
|
|
|
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)
|
|
|
|
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]
|
|
|
|
@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]
|
|
|
|
def get_bytes(self):
|
|
high = struct.pack("!Q", self.bits[1])
|
|
low = struct.pack("!Q", self.bits[0])
|
|
return high[-2:] + low
|
|
|
|
def get_bits(self):
|
|
return self.bits
|
|
|
|
def as_string(self) -> str:
|
|
return self.__str__()
|