diff --git a/README.rst b/README.rst index 7115d57..1ed809d 100644 --- a/README.rst +++ b/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 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..b210d91 --- /dev/null +++ b/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 diff --git a/tests/test_ulid.py b/tests/test_ulid.py index 11ae933..70edfe5 100644 --- a/tests/test_ulid.py +++ b/tests/test_ulid.py @@ -1,11 +1,13 @@ +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 @@ -13,17 +15,17 @@ 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 diff --git a/tests/value_provider/__init__.py b/tests/value_provider/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/value_provider/test_monotinoc_value_provider.py b/tests/value_provider/test_monotinoc_value_provider.py new file mode 100644 index 0000000..d7626da --- /dev/null +++ b/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() diff --git a/tests/value_provider/test_non_monotonic_value_provider.py b/tests/value_provider/test_non_monotonic_value_provider.py new file mode 100644 index 0000000..347a132 --- /dev/null +++ b/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 diff --git a/tests/value_provider/test_timestamp_value_base_provider.py b/tests/value_provider/test_timestamp_value_base_provider.py new file mode 100644 index 0000000..b9586e6 --- /dev/null +++ b/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) diff --git a/ulid/__init__.py b/ulid/__init__.py index d327385..01cb255 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -1,27 +1,22 @@ from __future__ import annotations import functools -import os -import time import uuid from datetime import datetime from datetime import timezone -from threading import Lock from typing import Any from typing import cast -from typing import Generic from typing import TYPE_CHECKING -from typing import TypeVar from typing_extensions import Self from ulid import base32 from ulid import constants +from ulid.value_provider import AbstractValueProvider +from ulid.value_provider import MonotonicValueProvider as ValueProvider if TYPE_CHECKING: # pragma: no cover - from collections.abc import Callable - from pydantic import GetCoreSchemaHandler from pydantic import ValidatorFunctionWrapHandler from pydantic_core import CoreSchema @@ -34,58 +29,15 @@ __version__ = version("python-ulid") -T = TypeVar("T", bound=type) -R = TypeVar("R") - - -class validate_type(Generic[T]): # noqa: N801 - def __init__(self, *types: T) -> None: - self.types = types - - def __call__(self, func: Callable[..., R]) -> Callable[..., R]: - @functools.wraps(func) - def wrapped(cls: Any, value: T) -> R: - if not isinstance(value, self.types): - message = "Value has to be of type " - message += " or ".join([t.__name__ for t in self.types]) - raise TypeError(message) - return func(cls, value) - - return wrapped - - -class ValueProvider: - def __init__(self) -> None: - self.lock = Lock() - self.prev_timestamp = constants.MIN_TIMESTAMP - self.prev_randomness = constants.MIN_RANDOMNESS - - def timestamp(self, value: float | None = None) -> int: - if value is None: - value = time.time_ns() // constants.NANOSECS_IN_MILLISECS - elif isinstance(value, float): - value = int(value * constants.MILLISECS_IN_SECS) - if value > constants.MAX_TIMESTAMP: - raise ValueError("Value exceeds maximum possible timestamp") - return value - - def randomness(self) -> bytes: - with self.lock: - current_timestamp = self.timestamp() - if current_timestamp == self.prev_timestamp: - if self.prev_randomness == constants.MAX_RANDOMNESS: - raise ValueError("Randomness within same millisecond exhausted") - randomness = self.increment_bytes(self.prev_randomness) - else: - randomness = os.urandom(constants.RANDOMNESS_LEN) - - self.prev_randomness = randomness - self.prev_timestamp = current_timestamp - return randomness - def increment_bytes(self, value: bytes) -> bytes: - length = len(value) - return (int.from_bytes(value, byteorder="big") + 1).to_bytes(length, byteorder="big") +def validate_value_type( + value_to_validate: object, + *types_to_validate_against: type, +) -> None: + if not isinstance(value_to_validate, types_to_validate_against): + message = "Value has to be of type " + message += " or ".join([t.__name__ for t in types_to_validate_against]) + raise TypeError(message) @functools.total_ordering @@ -109,23 +61,43 @@ class ULID: >>> str(ulid) '01E75PVKXA3GFABX1M1J9NZZNF' + The value provider will be used to generate the randomness part + (and the timestamp part if needed) of the `ULID`. + Args: value (bytes, None): A sequence of 16 bytes representing an encoded ULID. + value_provider (AbstractValueProvider, None): The value provider to use to generate the randomness and timestamp. Raises: ValueError: If the provided value is not a valid encoded ULID. """ - def __init__(self, value: bytes | None = None) -> None: + def __init__( + self, + value: bytes | None = None, + value_provider: AbstractValueProvider | None = None, + ) -> None: if value is not None and len(value) != constants.BYTES_LEN: raise ValueError("ULID has to be exactly 16 bytes long.") - self.bytes: bytes = value or ULID.from_timestamp(self.provider.timestamp()).bytes + value_provider_to_use = value_provider or self.provider + self.bytes: bytes = ( + value + or ULID.from_timestamp( + value_provider_to_use.timestamp(), value_provider=value_provider_to_use + ).bytes + ) @classmethod - @validate_type(datetime) - def from_datetime(cls, value: datetime) -> Self: + def from_datetime( + cls, + value: datetime, + value_provider: AbstractValueProvider | None = None, + ) -> Self: """Create a new :class:`ULID`-object from a :class:`datetime`. The timestamp part of the `ULID` will be set to the corresponding timestamp of the datetime. + The value provider will be used to + generate the randomness part of the + `ULID`. Examples: @@ -133,14 +105,22 @@ def from_datetime(cls, value: datetime) -> Self: >>> ULID.from_datetime(datetime.now()) ULID(01E75QRYCAMM1MKQ9NYMYT6SAV) """ - return cls.from_timestamp(value.timestamp()) + value_provider_to_use = value_provider or cls.provider + validate_value_type(value, datetime) + return cls.from_timestamp(value.timestamp(), value_provider=value_provider_to_use) @classmethod - @validate_type(int, float) - def from_timestamp(cls, value: float) -> Self: + def from_timestamp( + cls, + value: float, + value_provider: AbstractValueProvider | None = None, + ) -> Self: """Create a new :class:`ULID`-object from a timestamp. The timestamp can be either a `float` representing the time in seconds (as it would be returned by :func:`time.time()`) or an `int` in milliseconds. + The value provider will be used to + generate the randomness part of the + `ULID`. Examples: @@ -148,12 +128,15 @@ def from_timestamp(cls, value: float) -> Self: >>> ULID.from_timestamp(time.time()) ULID(01E75QWN5HKQ0JAVX9FG1K4YP4) """ - timestamp = int.to_bytes(cls.provider.timestamp(value), constants.TIMESTAMP_LEN, "big") - randomness = cls.provider.randomness() + validate_value_type(value, int, float) + value_provider_to_use = value_provider or cls.provider + timestamp = int.to_bytes( + value_provider_to_use.timestamp(value), constants.TIMESTAMP_LEN, "big" + ) + randomness = value_provider_to_use.randomness() return cls.from_bytes(timestamp + randomness) @classmethod - @validate_type(uuid.UUID) def from_uuid(cls, value: uuid.UUID) -> Self: """Create a new :class:`ULID`-object from a :class:`uuid.UUID`. The timestamp part will be random in that case. @@ -164,30 +147,31 @@ def from_uuid(cls, value: uuid.UUID) -> Self: >>> ULID.from_uuid(uuid4()) ULID(27Q506DP7E9YNRXA0XVD8Z5YSG) """ + validate_value_type(value, uuid.UUID) return cls(value.bytes) @classmethod - @validate_type(bytes) def from_bytes(cls, bytes_: bytes) -> Self: """Create a new :class:`ULID`-object from sequence of 16 bytes.""" + validate_value_type(bytes_, bytes) return cls(bytes_) @classmethod - @validate_type(str) def from_hex(cls, value: str) -> Self: """Create a new :class:`ULID`-object from 32 character string of hex values.""" + validate_value_type(value, str) return cls.from_bytes(bytes.fromhex(value)) @classmethod - @validate_type(str) def from_str(cls, string: str) -> Self: """Create a new :class:`ULID`-object from a 26 char long string representation.""" + validate_value_type(string, str) return cls(base32.decode(string)) @classmethod - @validate_type(int) def from_int(cls, value: int) -> Self: """Create a new :class:`ULID`-object from an `int`.""" + validate_value_type(value, int) return cls(int.to_bytes(value, constants.BYTES_LEN, "big")) @classmethod diff --git a/ulid/value_provider/__init__.py b/ulid/value_provider/__init__.py new file mode 100644 index 0000000..5e00b5f --- /dev/null +++ b/ulid/value_provider/__init__.py @@ -0,0 +1,10 @@ +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 + + +__all__ = [ + "AbstractValueProvider", + "MonotonicValueProvider", + "NonMonotonicValueProvider", +] diff --git a/ulid/value_provider/abstract_value_provider.py b/ulid/value_provider/abstract_value_provider.py new file mode 100644 index 0000000..d238fbd --- /dev/null +++ b/ulid/value_provider/abstract_value_provider.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import time +from abc import ABC +from abc import abstractmethod + +from ulid import constants + + +class AbstractValueProvider(ABC): + def timestamp(self, value: float | None = None) -> int: + """ + Generate a timestamp value. + Uses current time in milliseconds if no value is provided, + otherwise converts the provided timestamp in seconds to milliseconds. + """ + if value is None: + value = time.time_ns() // constants.NANOSECS_IN_MILLISECS + elif isinstance(value, float): + value = int(value * constants.MILLISECS_IN_SECS) + if value > constants.MAX_TIMESTAMP: + raise ValueError("Value exceeds maximum possible timestamp") + return value + + @abstractmethod + def randomness(self) -> bytes: + """ + Generate the randomness value. + """ diff --git a/ulid/value_provider/monotonic_value_provider.py b/ulid/value_provider/monotonic_value_provider.py new file mode 100644 index 0000000..f60e1de --- /dev/null +++ b/ulid/value_provider/monotonic_value_provider.py @@ -0,0 +1,30 @@ +import os +from threading import Lock + +from ulid import constants +from ulid.value_provider.abstract_value_provider import AbstractValueProvider + + +class MonotonicValueProvider(AbstractValueProvider): + def __init__(self) -> None: + self.lock = Lock() + self.prev_timestamp = constants.MIN_TIMESTAMP + self.prev_randomness = constants.MIN_RANDOMNESS + + def randomness(self) -> bytes: + with self.lock: + current_timestamp = self.timestamp() + if current_timestamp == self.prev_timestamp: + if self.prev_randomness == constants.MAX_RANDOMNESS: + raise ValueError("Randomness within same millisecond exhausted") + randomness = self.increment_bytes(self.prev_randomness) + else: + randomness = os.urandom(constants.RANDOMNESS_LEN) + + self.prev_randomness = randomness + self.prev_timestamp = current_timestamp + return randomness + + def increment_bytes(self, value: bytes) -> bytes: + length = len(value) + return (int.from_bytes(value, byteorder="big") + 1).to_bytes(length, byteorder="big") diff --git a/ulid/value_provider/non_monotonic_value_provider.py b/ulid/value_provider/non_monotonic_value_provider.py new file mode 100644 index 0000000..1b106ae --- /dev/null +++ b/ulid/value_provider/non_monotonic_value_provider.py @@ -0,0 +1,9 @@ +import os + +from ulid import constants +from ulid.value_provider.abstract_value_provider import AbstractValueProvider + + +class NonMonotonicValueProvider(AbstractValueProvider): + def randomness(self) -> bytes: + return os.urandom(constants.RANDOMNESS_LEN)