Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -139,8 +139,8 @@ components = generator.extract_snowflake_info(snowflake_id)
print(components)
```

This will return a dictionary with the following keys:
- `timestamp`: The time at which the ID was generated, formatted as a human-readable string.
This will return a SnowflakeIDInfo with the following attributes:
- `datetime`: The time at which the ID was generated, datetime class.
- `node_id`: The ID of the node that generated the ID.
- `worker_id`: The ID of the worker that generated the ID.
- `sequence`: The sequence number within the same millisecond.
Expand Down
37 changes: 26 additions & 11 deletions snowflakeid/generator.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,28 @@
import asyncio
import time
from dataclasses import dataclass
import datetime as dt
from typing import Optional, Dict

# Constants
DEFAULT_EPOCH_MS = 1723323246031
DEFAULT_EPOCH = 1735689600000
BASE62_CHARS = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
BASE62_BASE = len(BASE62_CHARS)


@dataclass(frozen=True)
class SnowflakeIDInfo:
datetime: dt.datetime
node_id: int
worker_id: int
sequence: int



@dataclass(frozen=True)
class SnowflakeIDConfig:
"""Configuration for the Snowflake ID generator."""
epoch: int = None
epoch: int = DEFAULT_EPOCH
total_bits: int = 64
time_bits: int = 39
node_bits: int = 7
Expand Down Expand Up @@ -109,7 +119,7 @@ def decode_base62(encoded_id: str) -> int:
decoded += BASE62_CHARS.index(char) * (BASE62_BASE ** i)
return decoded

def extract_snowflake_info(self, snowflake_id: int) -> Dict[str, int]:
def extract_snowflake_info(self, snowflake_id: int) -> SnowflakeIDInfo:
"""Extracts the components of a Snowflake ID.

Returns:
Expand All @@ -135,11 +145,16 @@ def extract_snowflake_info(self, snowflake_id: int) -> Dict[str, int]:
sequence = snowflake_id & sequence_mask

# parse timestamp to readable format
timestamp = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(timestamp / 1000))

return {
"timestamp": timestamp,
"worker_id": worker_id,
"node_id": node_id,
"sequence": sequence,
}
# timestamp = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(timestamp / 1000))
return SnowflakeIDInfo(
datetime=dt.datetime.fromtimestamp(timestamp / 1000),
worker_id=worker_id,
node_id=node_id,
sequence=sequence
)
# return {
# "timestamp": timestamp,
# "worker_id": worker_id,
# "node_id": node_id,
# "sequence": sequence,
# }
15 changes: 8 additions & 7 deletions tests/test_32_bits.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@

import pytest

from snowflakeid.snowflakeid import SnowflakeIDGenerator, SnowflakeIDConfig
from snowflakeid import SnowflakeIDGenerator, SnowflakeIDConfig

from snowflakeid.generator import SnowflakeIDInfo

# Define 32-bit configuration for testing
TEST_CONFIG_32BIT = SnowflakeIDConfig(
Expand Down Expand Up @@ -117,13 +119,12 @@ async def test_extract_snowflake_info_32bit():
"""Test extracting information from a 32-bit Snowflake ID."""
generator = SnowflakeIDGenerator(config=TEST_CONFIG_32BIT)
snowflake_id = await generator.generate()
info = generator.extract_snowflake_info(snowflake_id)
print(info)
info: SnowflakeIDInfo = generator.extract_snowflake_info(snowflake_id)

assert info["timestamp"] is not None, "Timestamp should be extracted."
assert info["worker_id"] == TEST_CONFIG_32BIT.worker_id, "Incorrect worker ID extracted."
assert info["node_id"] == TEST_CONFIG_32BIT.node_id, "Incorrect node ID extracted."
assert info["sequence"] >= 0, "Sequence should be a non-negative integer."
assert info.datetime is not None, "Timestamp should be extracted."
assert info.worker_id == TEST_CONFIG_32BIT.worker_id, "Incorrect worker ID extracted."
assert info.node_id == TEST_CONFIG_32BIT.node_id, "Incorrect node ID extracted."
assert info.sequence >= 0, "Sequence should be a non-negative integer."


# # Intentionally create a collision scenario for testing purposes
Expand Down
15 changes: 8 additions & 7 deletions tests/test_64_bits.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@

import pytest

from snowflakeid.snowflakeid import SnowflakeIDGenerator, SnowflakeIDConfig
from snowflakeid import SnowflakeIDGenerator, SnowflakeIDConfig
from snowflakeid.generator import SnowflakeIDInfo

# Define 64-bit configuration for testing
TEST_CONFIG_64BIT = SnowflakeIDConfig(
Expand Down Expand Up @@ -118,12 +119,12 @@ async def test_extract_snowflake_info_32bit():
"""Test extracting information from a 32-bit Snowflake ID."""
generator = SnowflakeIDGenerator(config=TEST_CONFIG_64BIT)
snowflake_id = await generator.generate()
info = generator.extract_snowflake_info(snowflake_id)
info: SnowflakeIDInfo = generator.extract_snowflake_info(snowflake_id)

assert info["timestamp"] is not None, "Timestamp should be extracted."
assert info["worker_id"] == TEST_CONFIG_64BIT.worker_id, "Incorrect worker ID extracted."
assert info["node_id"] == TEST_CONFIG_64BIT.node_id, "Incorrect node ID extracted."
assert info["sequence"] >= 0, "Sequence should be a non-negative integer."
assert info.datetime is not None, "Timestamp should be extracted."
assert info.worker_id == TEST_CONFIG_64BIT.worker_id, "Incorrect worker ID extracted."
assert info.node_id == TEST_CONFIG_64BIT.node_id, "Incorrect node ID extracted."
assert info.sequence >= 0, "Sequence should be a non-negative integer."


# # Intentionally create a collision scenario for testing purposes
Expand All @@ -141,4 +142,4 @@ async def test_intentional_collision_32bit():
id2 = await generator2.generate()

with pytest.raises(AssertionError):
assert id1 == id2, "Intentional collision failed."
assert id1 == id2, "Intentional collision failed."