Add Non monotonic support

README.rst

@@ -153,8 +153,8 @@ The ``ULID`` class can be directly used for the popular data validation library
 
 .. monotonic-begin
 
-Monotonic Support
------------------
+Monotonic And Non-Monotonic Support
+-----------------------------------
 
 This library by default supports the implementation for monotonic sort order suggested by the
 official ULID specification.
@@ -163,6 +163,39 @@ This means that ULID values generated in the same millisecond will have linear i
 values. If :math:`r_1` and :math:`r_2` are the randomness values of two ULIDs with the same
 timestamp, then :math:`r_2 = r_1 + 1`.
 
+You can override this implementation by providing your own value provider to the ``ULID``
+constructor. The library comes with a default monotonic (``MonotonicValueProvider``)
+and non-monotonic (``NonMonotonicValueProvider``) value provider that generates
+randomness values independently. For example:
+
+.. code-block:: python
+
+  from ulid import ULID
+  from ulid.value_provider import NonMonotonicValueProvider
+
+  ulid1 = ULID(value_provider=NonMonotonicValueProvider())
+  ulid2 = ULID(value_provider=NonMonotonicValueProvider())
+
+You can also implement your own value provider by inheriting from the ``AbstractValueProvider`` and
+overriding the ``randomness`` and ``timestamp`` methods. For example:
+
+.. code-block:: python
+
+  from ulid import ULID
+  from ulid.value_provider import AbstractValueProvider
+
+  class MyValueProvider(AbstractValueProvider):
+      def randomness(self) -> bytes:
+          # Implement your randomness generation logic here
+          return b"\x00" * 10  # Example: return a fixed randomness value
+
+      def timestamp(self, value: float | None = None) -> int:
+          # Implement your timestamp generation logic here
+          return 1772790331000  # Example: return a fixed timestamp value
+
+   ulid1 = ULID(value_provider=MyValueProvider())
+   print(ulid1) # Should print "ULID(01KK18KDKR0000000000000000)"
+
 .. monotonic-end
 
 .. cli-begin

tests/conftest.py

@@ -0,0 +1,17 @@
+from datetime import datetime
+from datetime import timezone
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def datetimes_almost_equal(a: datetime, b: datetime) -> None:
+    assert a.replace(microsecond=0) == b.replace(microsecond=0)
+
+
+def assert_sorted(seq: list) -> None:
+    last = seq[0]
+    for item in seq[1:]:
+        assert last < item
+        last = item

tests/test_ulid.py

@@ -1,29 +1,31 @@
+from __future__ import annotations
+
 import json
 import time
 import uuid
-from collections.abc import Callable
 from datetime import datetime
 from datetime import timedelta
 from datetime import timezone
 from typing import Optional
+from typing import TYPE_CHECKING
 from typing import Union
 
 import pytest
 from freezegun import freeze_time
 from pydantic import BaseModel
 from pydantic import ValidationError
 
+from tests.conftest import assert_sorted
+from tests.conftest import datetimes_almost_equal
+from tests.conftest import utcnow
 from ulid import base32
 from ulid import constants
 from ulid import ULID
+from ulid.value_provider.abstract_value_provider import AbstractValueProvider
 
 
-def utcnow() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def datetimes_almost_equal(a: datetime, b: datetime) -> None:
-    assert a.replace(microsecond=0) == b.replace(microsecond=0)
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 
 @freeze_time()
@@ -70,17 +72,11 @@ def test_same_millisecond_monotonic_sorting() -> None:
 @freeze_time()
 def test_same_millisecond_overflow() -> None:
     ULID.provider.prev_randomness = constants.MAX_RANDOMNESS
+    ULID.provider.prev_timestamp = ULID.provider.timestamp()
     with pytest.raises(ValueError, match="Randomness within same millisecond exhausted"):
         ULID()
 
 
-def assert_sorted(seq: list) -> None:
-    last = seq[0]
-    for item in seq[1:]:
-        assert last < item
-        last = item
-
-
 def test_comparison() -> None:
     with freeze_time() as frozen_time:
         ulid1 = ULID()
@@ -233,7 +229,7 @@ def test_pydantic_protocol() -> None:
     ulid = ULID()
 
     class Model(BaseModel):
-        ulid: Optional[ULID] = None  # noqa: FA100
+        ulid: Optional[ULID] = None
 
     model: Model | None = None
     for value in [ulid, str(ulid), int(ulid), bytes(ulid)]:
@@ -275,3 +271,93 @@ class Model(BaseModel):
     assert {
         "type": "null",
     } in model_json_schema["properties"]["ulid"]["anyOf"]
+
+
+def test_ulid_constructor_support_other_value_provider() -> None:
+    random_part = b"\x00" * 10
+    datetime = utcnow()
+    timestamp_in_seconds = int(datetime.timestamp())
+    timestamp_in_milliseconds = int(timestamp_in_seconds * constants.MILLISECS_IN_SECS)
+    ulid_bytes: bytes = (
+        timestamp_in_milliseconds.to_bytes(constants.TIMESTAMP_LEN, byteorder="big") + random_part
+    )
+
+    class DummyValueProvider(AbstractValueProvider):
+        def randomness(self) -> bytes:
+            return random_part
+
+        def timestamp(self, value: float | None = None) -> int:  # noqa: ARG002 because we overriding but still don't have `typing_extensions`.
+            return timestamp_in_milliseconds
+
+    ulid1 = ULID(value_provider=DummyValueProvider())
+    ulid2 = ULID(value_provider=DummyValueProvider())
+
+    assert ulid1.bytes == ulid_bytes
+    assert ulid1.timestamp == timestamp_in_seconds
+    datetimes_almost_equal(ulid1.datetime, datetime)
+    assert ulid1.milliseconds == timestamp_in_milliseconds
+    assert ulid1.hex == ulid_bytes.hex()
+    assert str(ulid1) == base32.encode(ulid_bytes)
+    # since the same dummy value provider is used,
+    # the generated ULIDs should be the same.
+    assert ulid2 == ulid1
+
+
+def test_ulid_from_datetime_support_other_value_provider() -> None:
+    random_part = b"\x00" * 10
+    datetime = utcnow()
+    timestamp_in_seconds = int(datetime.timestamp())
+    timestamp_in_milliseconds = int(timestamp_in_seconds * constants.MILLISECS_IN_SECS)
+    ulid_bytes: bytes = (
+        timestamp_in_milliseconds.to_bytes(constants.TIMESTAMP_LEN, byteorder="big") + random_part
+    )
+
+    class DummyValueProvider(AbstractValueProvider):
+        def randomness(self) -> bytes:
+            return random_part
+
+        def timestamp(self, value: float | None = None) -> int:  # noqa: ARG002 because we overriding but still don't have `typing_extensions`.
+            return timestamp_in_milliseconds
+
+    ulid1 = ULID.from_datetime(datetime, value_provider=DummyValueProvider())
+    ulid2 = ULID.from_datetime(datetime, value_provider=DummyValueProvider())
+
+    assert ulid1.bytes == ulid_bytes
+    assert ulid1.timestamp == timestamp_in_seconds
+    datetimes_almost_equal(ulid1.datetime, datetime)
+    assert ulid1.milliseconds == timestamp_in_milliseconds
+    assert ulid1.hex == ulid_bytes.hex()
+    assert str(ulid1) == base32.encode(ulid_bytes)
+    # since the same dummy value provider is used,
+    # the generated ULIDs should be the same.
+    assert ulid2 == ulid1
+
+
+def test_ulid_from_timestamp_support_other_value_provider() -> None:
+    random_part = b"\x00" * 10
+    datetime = utcnow()
+    timestamp_in_seconds = int(datetime.timestamp())
+    timestamp_in_milliseconds = int(timestamp_in_seconds * constants.MILLISECS_IN_SECS)
+    ulid_bytes: bytes = (
+        timestamp_in_milliseconds.to_bytes(constants.TIMESTAMP_LEN, byteorder="big") + random_part
+    )
+
+    class DummyValueProvider(AbstractValueProvider):
+        def randomness(self) -> bytes:
+            return random_part
+
+        def timestamp(self, value: float | None = None) -> int:  # noqa: ARG002 because we overriding but still don't have `typing_extensions`.
+            return timestamp_in_milliseconds
+
+    ulid1 = ULID.from_timestamp(datetime.timestamp(), value_provider=DummyValueProvider())
+    ulid2 = ULID.from_timestamp(datetime.timestamp(), value_provider=DummyValueProvider())
+
+    assert ulid1.bytes == ulid_bytes
+    assert ulid1.timestamp == timestamp_in_seconds
+    datetimes_almost_equal(ulid1.datetime, datetime)
+    assert ulid1.milliseconds == timestamp_in_milliseconds
+    assert ulid1.hex == ulid_bytes.hex()
+    assert str(ulid1) == base32.encode(ulid_bytes)
+    # since the same dummy value provider is used,
+    # the generated ULIDs should be the same.
+    assert ulid2 == ulid1

tests/value_provider/test_monotinoc_value_provider.py

@@ -0,0 +1,36 @@
+import pytest
+from freezegun import freeze_time
+
+from ulid import constants
+from ulid.value_provider import MonotonicValueProvider
+
+
+def test_generate_randomness_monotinic() -> None:
+    provider = MonotonicValueProvider()
+
+    randomness1: bytes = provider.randomness()
+    randomness1_as_number = int.from_bytes(randomness1, byteorder="big")
+    randomness2: bytes = provider.randomness()
+    randomness2_as_number = int.from_bytes(randomness2, byteorder="big")
+    randomness3: bytes = provider.randomness()
+    randomness3_as_number = int.from_bytes(randomness3, byteorder="big")
+
+    assert len(randomness1) == len(randomness2) == len(randomness3) == constants.RANDOMNESS_LEN
+    # Assert that they are monotonic and not equal.
+    assert randomness2_as_number - randomness1_as_number == 1
+    assert randomness3_as_number - randomness2_as_number == 1
+    assert randomness3_as_number - randomness1_as_number == 2  # noqa: PLR2004 Allow use of magic numbers.
+
+
+def test_randomness_exhaustion() -> None:
+    provider = MonotonicValueProvider()
+
+    # Set the previous randomness to the maximum value.
+    provider.prev_randomness = constants.MAX_RANDOMNESS
+
+    # Attempting to generate randomness within the
+    # same millisecond should raise an error.
+    with freeze_time():
+        provider.prev_timestamp = provider.timestamp()
+        with pytest.raises(ValueError, match="Randomness within same millisecond exhausted"):
+            provider.randomness()

tests/value_provider/test_non_monotonic_value_provider.py

@@ -0,0 +1,15 @@
+from ulid import constants
+from ulid.value_provider import NonMonotonicValueProvider
+
+
+def test_generate_randomness() -> None:
+    provider = NonMonotonicValueProvider()
+
+    randomness1: bytes = provider.randomness()
+    randomness1_as_number = int.from_bytes(randomness1, byteorder="big")
+    randomness2: bytes = provider.randomness()
+    randomness2_as_number = int.from_bytes(randomness2, byteorder="big")
+
+    assert len(randomness1) == len(randomness2) == constants.RANDOMNESS_LEN
+    # Assert that they are not monotonic and not equal.
+    assert abs(randomness1_as_number - randomness2_as_number) > 1

tests/value_provider/test_timestamp_value_base_provider.py

@@ -0,0 +1,89 @@
+from datetime import datetime
+from datetime import timezone
+
+import pytest
+from freezegun import freeze_time
+
+from tests.conftest import utcnow
+from ulid import constants
+from ulid.value_provider.abstract_value_provider import AbstractValueProvider
+from ulid.value_provider.monotonic_value_provider import MonotonicValueProvider
+from ulid.value_provider.non_monotonic_value_provider import NonMonotonicValueProvider
+
+
+class TestValueProvider(AbstractValueProvider):
+    def randomness(self) -> bytes:
+        return b"\x00" * 10
+
+
+@pytest.mark.parametrize(
+    "datetime_timestamp",
+    [
+        datetime(2026, 6, 6, 6, 6, 6, 6, tzinfo=timezone.utc),
+        datetime(2024, 1, 1, tzinfo=timezone.utc),
+        datetime(2025, 12, 31, 9, 6, 3, tzinfo=timezone.utc),
+    ],
+)
+@pytest.mark.parametrize(
+    "value_provider",
+    [
+        pytest.param(TestValueProvider(), id="CustomedValueProvider"),
+        pytest.param(MonotonicValueProvider(), id="MonotonicValueProvider"),
+        pytest.param(NonMonotonicValueProvider(), id="NonMonotonicValueProvider"),
+    ],
+)
+def test_timestamp(
+    datetime_timestamp: datetime,
+    value_provider: AbstractValueProvider,
+) -> None:
+    expected_timestamp = int(datetime_timestamp.timestamp() * constants.MILLISECS_IN_SECS)
+
+    first_timestamp = value_provider.timestamp(datetime_timestamp.timestamp())
+    second_timestamp = value_provider.timestamp(datetime_timestamp.timestamp())
+
+    assert isinstance(first_timestamp, int)
+    assert first_timestamp == expected_timestamp
+    assert isinstance(second_timestamp, int)
+    assert second_timestamp == expected_timestamp
+    assert second_timestamp == first_timestamp
+
+
+@freeze_time()
+@pytest.mark.parametrize(
+    "value_provider",
+    [
+        pytest.param(TestValueProvider(), id="CustomedValueProvider"),
+        pytest.param(MonotonicValueProvider(), id="MonotonicValueProvider"),
+        pytest.param(NonMonotonicValueProvider(), id="NonMonotonicValueProvider"),
+    ],
+)
+def test_timestamp_now(
+    value_provider: AbstractValueProvider,
+) -> None:
+    with freeze_time() as frozen:
+        expected_first_timestamp = int(utcnow().timestamp() * constants.MILLISECS_IN_SECS)
+        first_timestamp = value_provider.timestamp()
+        frozen.tick()
+        expected_second_timestamp = int(utcnow().timestamp() * constants.MILLISECS_IN_SECS)
+        second_timestamp = value_provider.timestamp()
+
+    assert isinstance(first_timestamp, int)
+    assert first_timestamp == expected_first_timestamp
+    assert isinstance(second_timestamp, int)
+    assert second_timestamp == expected_second_timestamp
+    assert second_timestamp > first_timestamp
+
+
+@pytest.mark.parametrize(
+    "value_provider",
+    [
+        pytest.param(TestValueProvider(), id="CustomedValueProvider"),
+        pytest.param(MonotonicValueProvider(), id="MonotonicValueProvider"),
+        pytest.param(NonMonotonicValueProvider(), id="NonMonotonicValueProvider"),
+    ],
+)
+def test_max_timestamp(
+    value_provider: AbstractValueProvider,
+) -> None:
+    with pytest.raises(ValueError, match="Value exceeds maximum possible timestamp"):
+        value_provider.timestamp(constants.MAX_TIMESTAMP + 1)