From 14ca7667e24e3d82a7a938cc0bd9a97c4cb4f6dc Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 11:30:10 +0200 Subject: [PATCH 1/8] refactor(core): decouple monotonic generation state into independent ULIDGenerator Extracts the stateful monotonicity tracking, thread locks, and clock/randomness sourcing out of the global class-level scope and encapsulates it in a clean, isolated `ULIDGenerator` class. Why this change was made: 1. Global isolation: Avoids global mutable state pollution where calling ULID() affects the generation of all other ULIDs in different parts of an application. 2. Testability: Tests no longer need to mutate global class-level properties of `ULID.provider` directly. They can now cleanly instantiate isolated generators and inject mock clocks or mock randomness sources, aligning with the principle "the interface is the test surface". 3. Customizability: Callers can now configure and instantiate distinct generator instances with different lifecycles and behaviors. All backward compatibility is fully preserved. A default shared generator is created inside `ulid` so that calls to `ULID()` and `from_timestamp` continue working out-of-the-box. Added full unit tests to cover state isolation and custom clocks/entropy injection. --- CONTEXT.md | 31 +++++++++++++++++++++ tests/test_ulid.py | 38 ++++++++++++++++++++++++-- ulid/__init__.py | 68 ++++++++++++++++++++++++++++++++++------------ 3 files changed, 117 insertions(+), 20 deletions(-) create mode 100644 CONTEXT.md diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000..4fb1a4c --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,31 @@ +# Domain Glossary & Context: python-ulid + +Welcome to the architectural domain reference for the `python-ulid` project. This document outlines the ubiquitous language, core domain entities, and structural concepts of this implementation. + +--- + +## Core Concepts & Vocabulary + +### ULID +* **Definition**: A Universally Unique Lexicographically Sortable Identifier. +* **Format**: A 128-bit value consisting of: + * **Timestamp**: 48 bits, representing epoch time in milliseconds. + * **Randomness**: 80 bits, representing high-entropy random generation. +* **Representation**: Encoded as a 26-character Base32 string. +* **Nature**: Act as an immutable, hashable **Value Object**. + +### ULIDGenerator +* **Definition**: A deep, stateful generator responsible for orchestrating the creation of new `ULID` identifiers. +* **Responsibilities**: + * Sampling system or injected clocks for the **Timestamp**. + * Sourcing entropy for the **Randomness**. + * Tracking state and enforcing **Monotonicity** rules. + * Guaranteeing thread-safe generation across execution contexts. + +### Monotonicity +* **Definition**: The deterministic ordering property where multiple ULID instances generated within the exact same millisecond increment their randomness component by $1$ to prevent sorting collisions. +* **Friction**: Under heavy concurrent execution, generating more identifiers than the 80-bit randomness field allows in a single millisecond will raise an overflow error. + +### Base32 Engine +* **Definition**: Crockford's Base32 translation layer. +* **Nature**: Encodes/decodes binary representation using a restricted alphabet of 32 characters (`0123456789ABCDEFGHJKMNPQRSTVWXYZ`), omitting ambiguous characters like `I`, `L`, `O`, and `U` to guarantee maximum readability. diff --git a/tests/test_ulid.py b/tests/test_ulid.py index 0e14f8b..21d2942 100644 --- a/tests/test_ulid.py +++ b/tests/test_ulid.py @@ -15,6 +15,7 @@ from ulid import base32 from ulid import constants from ulid import ULID +from ulid import ULIDGenerator def utcnow() -> datetime: @@ -80,10 +81,41 @@ def test_z_real_time_monotonic_sorting() -> None: @freeze_time() def test_same_millisecond_overflow() -> None: - ULID.provider.prev_timestamp = ULID.provider.timestamp() - ULID.provider.prev_randomness = constants.MAX_RANDOMNESS + generator = ULIDGenerator(randomness=lambda _: constants.MAX_RANDOMNESS) + generator.generate() with pytest.raises(ValueError, match="Randomness within same millisecond exhausted"): - ULID() + generator.generate() + + +def test_generator_custom_clock_and_randomness() -> None: + custom_ts = 123456789 + custom_rand = b"1234567890" + generator = ULIDGenerator( + clock=lambda: custom_ts, + randomness=lambda _: custom_rand, + ) + ulid = generator.generate() + assert ulid.milliseconds == custom_ts + assert ulid.bytes[constants.TIMESTAMP_LEN :] == custom_rand + + +def test_generator_state_isolation() -> None: + # Ensure two generators do not share state + g1 = ULIDGenerator() + g2 = ULIDGenerator() + + # Initially, both states should be untouched + assert g1.prev_timestamp == constants.MIN_TIMESTAMP + assert g2.prev_timestamp == constants.MIN_TIMESTAMP + + # Generating with g1 should update g1's state, but g2 should remain untouched + g1.generate() + assert g1.prev_timestamp != constants.MIN_TIMESTAMP + assert g2.prev_timestamp == constants.MIN_TIMESTAMP + + # Generating with g2 should update g2's state independently + g2.generate() + assert g2.prev_timestamp != constants.MIN_TIMESTAMP def assert_sorted(seq: list[Any]) -> None: diff --git a/ulid/__init__.py b/ulid/__init__.py index 63c6682..99bafe2 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -58,45 +58,82 @@ def wrapped(cls: Any, value: T) -> R: return wrapped -class ValueProvider: - def __init__(self) -> None: +class ULIDGenerator: + """Generator for creating universally unique lexicographically sortable identifiers (ULIDs). + + This class handles stateful monotonic generation of ULIDs, ensuring that identifiers + generated within the same millisecond are monotonically increasing. + """ + + def __init__( + self, + clock: Callable[[], int] | None = None, + randomness: Callable[[int], bytes] | None = None, + ) -> None: self.lock = Lock() + self.clock = clock or self._default_clock + self.randomness_source = randomness or self._default_randomness self.prev_timestamp = constants.MIN_TIMESTAMP self.prev_randomness = constants.MIN_RANDOMNESS + @staticmethod + def _default_clock() -> int: + return time.time_ns() // constants.NANOSECS_IN_MILLISECS + + @staticmethod + def _default_randomness(_timestamp: int) -> bytes: + return os.urandom(constants.RANDOMNESS_LEN) + def timestamp(self, value: float | None = None) -> int: if value is None: - value = time.time_ns() // constants.NANOSECS_IN_MILLISECS + value = self.clock() 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, current_timestamp: int | None = None) -> bytes: + def generate(self, timestamp: float | datetime | None = None) -> ULID: + """Generate a new :class:`ULID` monotonically. + + Args: + timestamp (int, float, datetime, None): Optional timestamp to set on the ULID. + + Returns: + ULID: A generated ULID. + """ + ts_val: int + if isinstance(timestamp, datetime): + ts_val = self.timestamp(timestamp.timestamp()) + elif isinstance(timestamp, (int, float)): + ts_val = self.timestamp(timestamp) + else: + ts_val = self.timestamp() + with self.lock: - if current_timestamp is None: - current_timestamp = self.timestamp() - if current_timestamp == self.prev_timestamp: + if ts_val == 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) + randomness = self.randomness_source(ts_val) self.prev_randomness = randomness - self.prev_timestamp = current_timestamp - return randomness + self.prev_timestamp = ts_val + + timestamp_bytes = int.to_bytes(ts_val, constants.TIMESTAMP_LEN, "big") + return ULID.from_bytes(timestamp_bytes + randomness) def increment_bytes(self, value: bytes) -> bytes: length = len(value) return (int.from_bytes(value, byteorder="big") + 1).to_bytes(length, byteorder="big") +_default_generator = ULIDGenerator() + + @functools.total_ordering class ULID: - provider = ValueProvider() - """The :class:`ULID` object consists of a timestamp part of 48 bits and of 80 random bits. .. code-block:: text @@ -124,7 +161,7 @@ class ULID: def __init__(self, value: bytes | 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 + self.bytes: bytes = value or _default_generator.generate().bytes @classmethod @validate_type(datetime) @@ -153,10 +190,7 @@ def from_timestamp(cls, value: float) -> Self: >>> ULID.from_timestamp(time.time()) ULID(01E75QWN5HKQ0JAVX9FG1K4YP4) """ - timestamp_value = cls.provider.timestamp(value) - timestamp = int.to_bytes(timestamp_value, constants.TIMESTAMP_LEN, "big") - randomness = cls.provider.randomness(timestamp_value) - return cls.from_bytes(timestamp + randomness) + return cls.from_bytes(_default_generator.generate(value).bytes) @classmethod @validate_type(uuid.UUID) From bbbf8626d5bee2b9970cc1b7675d69d4a9c837be Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 11:40:05 +0200 Subject: [PATCH 2/8] refactor(core): streamline type validation and eliminate validate_type decorator Removes the obsolete `validate_type` generic class/decorator wrapper from all `ULID` class constructors and inlines clean, high-performance `isinstance` checks. Why this change was made: 1. Architectural depth: Replaced a shallow decorator module with direct, readable runtime type validation, satisfying "Proposal 2" of the architectural review. 2. Performance & Debuggability: Eliminates wrapper function overhead in critical paths and keeps stack traces clean during exception reporting. 3. Clean namespace: Safely removed obsolete `Generic` and `TypeVar` typing imports. Preserves exact backwards compatible TypeError messages for all validation paths. --- ulid/__init__.py | 45 ++++++++++++++++----------------------------- 1 file changed, 16 insertions(+), 29 deletions(-) diff --git a/ulid/__init__.py b/ulid/__init__.py index 99bafe2..46fa064 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -9,9 +9,7 @@ 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 ulid import base32 from ulid import constants @@ -38,25 +36,6 @@ __version__ = version("python-ulid") -T = TypeVar("T") -R = TypeVar("R") - - -class validate_type(Generic[T]): # noqa: N801 - def __init__(self, *types: type[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 ULIDGenerator: """Generator for creating universally unique lexicographically sortable identifiers (ULIDs). @@ -164,7 +143,6 @@ def __init__(self, value: bytes | None = None) -> None: self.bytes: bytes = value or _default_generator.generate().bytes @classmethod - @validate_type(datetime) def from_datetime(cls, value: datetime) -> 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. @@ -175,10 +153,11 @@ def from_datetime(cls, value: datetime) -> Self: >>> ULID.from_datetime(datetime.now()) ULID(01E75QRYCAMM1MKQ9NYMYT6SAV) """ + if not isinstance(value, datetime): + raise TypeError("Value has to be of type datetime") return cls.from_timestamp(value.timestamp()) @classmethod - @validate_type(int, float) def from_timestamp(cls, value: float) -> 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()`) @@ -190,10 +169,11 @@ def from_timestamp(cls, value: float) -> Self: >>> ULID.from_timestamp(time.time()) ULID(01E75QWN5HKQ0JAVX9FG1K4YP4) """ + if not isinstance(value, (int, float)): + raise TypeError("Value has to be of type int or float") return cls.from_bytes(_default_generator.generate(value).bytes) @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. @@ -204,30 +184,36 @@ def from_uuid(cls, value: uuid.UUID) -> Self: >>> ULID.from_uuid(uuid4()) ULID(27Q506DP7E9YNRXA0XVD8Z5YSG) """ + if not isinstance(value, uuid.UUID): + raise TypeError("Value has to be of type 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.""" + if not isinstance(bytes_, bytes): + raise TypeError("Value has to be of type 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.""" + if not isinstance(value, str): + raise TypeError("Value has to be of type 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.""" + if not isinstance(string, str): + raise TypeError("Value has to be of type 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`.""" + if not isinstance(value, int): + raise TypeError("Value has to be of type int") return cls(int.to_bytes(value, constants.BYTES_LEN, "big")) @classmethod @@ -368,7 +354,6 @@ def to_uuid7(self, *, compliant: bool = False) -> uuid.UUID: return uuid.UUID(bytes=uuid_bytes) @classmethod - @validate_type(uuid.UUID) def from_uuidv7(cls, value: uuid.UUID) -> Self: """Create a new :class:`ULID` from a UUIDv7 (:class:`uuid.UUID` version 7). @@ -383,6 +368,8 @@ def from_uuidv7(cls, value: uuid.UUID) -> Self: >>> ulid.datetime datetime.datetime(2025, 11, 10, ...) """ + if not isinstance(value, uuid.UUID): + raise TypeError("Value has to be of type UUID") uuid_int = int.from_bytes(value.bytes, byteorder="big") # Extract timestamp from UUIDv7 layout (always in first 48 bits) From 4c77a40c77252cac66a3f389bb7f69debca30b53 Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 12:03:04 +0200 Subject: [PATCH 3/8] feat(core): introduce extensible monotonicity policies for ULIDGenerator Introduces a swappable `MonotonicityPolicy` interface (Protocol) to configure and vary the generator's behavior during same-millisecond collisions. Implementations added: 1. `StrictMonotonicPolicy`: Always increments the randomness by 1 (default), raising ValueError if the 80-bit randomness is exhausted. 2. `PureRandomPolicy`: Completely ignores monotonicity and always returns fresh, high-entropy random bytes, maximizing security and unpredictability. 3. `LaxMonotonicPolicy`: Attempts to increment monotonically, but on randomness exhaustion, regenerates fresh randomness instead of raising errors or sleeping. Other Changes: - Exposes direct type-assertion checks in the test suite to safely verify isolated states. - Appended robust unit tests verifying the correctness of `PureRandomPolicy` and `LaxMonotonicPolicy` under collision and overflow constraints. - Updated `CONTEXT.md` to formally document swappable Monotonicity Policies. --- CONTEXT.md | 9 ++-- tests/test_ulid.py | 59 +++++++++++++++++++++-- ulid/__init__.py | 116 ++++++++++++++++++++++++++++++++++++++++----- 3 files changed, 165 insertions(+), 19 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index 4fb1a4c..97f87f3 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -22,9 +22,12 @@ Welcome to the architectural domain reference for the `python-ulid` project. Thi * Tracking state and enforcing **Monotonicity** rules. * Guaranteeing thread-safe generation across execution contexts. -### Monotonicity -* **Definition**: The deterministic ordering property where multiple ULID instances generated within the exact same millisecond increment their randomness component by $1$ to prevent sorting collisions. -* **Friction**: Under heavy concurrent execution, generating more identifiers than the 80-bit randomness field allows in a single millisecond will raise an overflow error. +### Monotonicity & Policies +* **Definition**: The deterministic ordering property where multiple ULID instances generated within the exact same millisecond resolve their randomness component to prevent sorting collisions. +* **MonotonicityPolicy**: An extensible interface (seam) for configuring generator behavior: + * **StrictMonotonicPolicy**: Always increments the randomness by 1 under same-millisecond collisions, raising an overflow error if randomness is exhausted. + * **PureRandomPolicy**: Ignores previous states and always generates fresh random bytes, maximizing security and entropy. + * **LaxMonotonicPolicy**: Increments monotonically, but on same-millisecond randomness exhaustion, regenerates fresh randomness instead of raising an error or sleeping. ### Base32 Engine * **Definition**: Crockford's Base32 translation layer. diff --git a/tests/test_ulid.py b/tests/test_ulid.py index 21d2942..ba13474 100644 --- a/tests/test_ulid.py +++ b/tests/test_ulid.py @@ -14,6 +14,9 @@ from ulid import base32 from ulid import constants +from ulid import LaxMonotonicPolicy +from ulid import PureRandomPolicy +from ulid import StrictMonotonicPolicy from ulid import ULID from ulid import ULIDGenerator @@ -104,18 +107,21 @@ def test_generator_state_isolation() -> None: g1 = ULIDGenerator() g2 = ULIDGenerator() + assert isinstance(g1.policy, StrictMonotonicPolicy) + assert isinstance(g2.policy, StrictMonotonicPolicy) + # Initially, both states should be untouched - assert g1.prev_timestamp == constants.MIN_TIMESTAMP - assert g2.prev_timestamp == constants.MIN_TIMESTAMP + assert g1.policy.prev_timestamp == constants.MIN_TIMESTAMP + assert g2.policy.prev_timestamp == constants.MIN_TIMESTAMP # Generating with g1 should update g1's state, but g2 should remain untouched g1.generate() - assert g1.prev_timestamp != constants.MIN_TIMESTAMP - assert g2.prev_timestamp == constants.MIN_TIMESTAMP + assert g1.policy.prev_timestamp != constants.MIN_TIMESTAMP + assert g2.policy.prev_timestamp == constants.MIN_TIMESTAMP # Generating with g2 should update g2's state independently g2.generate() - assert g2.prev_timestamp != constants.MIN_TIMESTAMP + assert g2.policy.prev_timestamp != constants.MIN_TIMESTAMP def assert_sorted(seq: list[Any]) -> None: @@ -432,3 +438,46 @@ class Model(BaseModel): assert { "type": "null", } in model_json_schema["properties"]["ulid"]["anyOf"] + + +def test_pure_random_policy() -> None: + # Ensure PureRandomPolicy generates non-monotonic fresh randomness + custom_ts = 123456789 + rand1 = b"1" * constants.RANDOMNESS_LEN + rand2 = b"2" * constants.RANDOMNESS_LEN + rands = [rand1, rand2] + + # Iterator to return our mocks + iterator = iter(rands) + generator = ULIDGenerator( + clock=lambda: custom_ts, + randomness=lambda _: next(iterator), + policy=PureRandomPolicy(), + ) + + ulid1 = generator.generate() + ulid2 = generator.generate() + + assert ulid1.bytes[constants.TIMESTAMP_LEN :] == rand1 + assert ulid2.bytes[constants.TIMESTAMP_LEN :] == rand2 + + +def test_lax_monotonic_policy() -> None: + # Under LaxMonotonicPolicy, normal operations increase monotonically + custom_ts = 123456789 + generator = ULIDGenerator( + clock=lambda: custom_ts, + policy=LaxMonotonicPolicy(), + ) + ulid1 = generator.generate() + ulid2 = generator.generate() + assert ulid1 < ulid2 + + # Mocking same-millisecond overflow + # Set the state of the lax policy to MAX_RANDOMNESS + assert isinstance(generator.policy, LaxMonotonicPolicy) + generator.policy.prev_randomness = constants.MAX_RANDOMNESS + + # Generating again should regenerate fresh randomness instead of raising ValueError or sleeping + ulid3 = generator.generate() + assert ulid3 is not None diff --git a/ulid/__init__.py b/ulid/__init__.py index 46fa064..3eb866e 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -9,6 +9,7 @@ from threading import Lock from typing import Any from typing import cast +from typing import Protocol from typing import TYPE_CHECKING from ulid import base32 @@ -37,6 +38,103 @@ __version__ = version("python-ulid") +class MonotonicityPolicy(Protocol): + """Protocol defining the interface for monotonicity and randomness resolution policies.""" + + def resolve_randomness( + self, + timestamp: int, + randomness_source: Callable[[int], bytes], + increment_bytes: Callable[[bytes], bytes], + ) -> bytes: + """Resolve randomness for a given timestamp. + + Args: + timestamp (int): The current timestamp in milliseconds. + randomness_source (Callable[[int], bytes]): A callable to get fresh random bytes. + increment_bytes (Callable[[bytes], bytes]): A callable to increment random bytes. + + Returns: + bytes: The resolved randomness bytes (80 bits). + """ + ... + + +class StrictMonotonicPolicy: + """Strict monotonicity policy. + + Always increments the randomness by 1 if generated within the same millisecond. + Raises ValueError on millisecond randomness exhaustion. + """ + + def __init__(self) -> None: + self.prev_timestamp = constants.MIN_TIMESTAMP + self.prev_randomness = constants.MIN_RANDOMNESS + + def resolve_randomness( + self, + timestamp: int, + randomness_source: Callable[[int], bytes], + increment_bytes: Callable[[bytes], bytes], + ) -> bytes: + if timestamp == self.prev_timestamp: + if self.prev_randomness == constants.MAX_RANDOMNESS: + raise ValueError("Randomness within same millisecond exhausted") + randomness = increment_bytes(self.prev_randomness) + else: + randomness = randomness_source(timestamp) + + self.prev_randomness = randomness + self.prev_timestamp = timestamp + return randomness + + +class PureRandomPolicy: + """Pure random policy. + + Always generates fresh randomness without enforcing any monotonicity constraints. + """ + + def resolve_randomness( + self, + timestamp: int, + randomness_source: Callable[[int], bytes], + increment_bytes: Callable[[bytes], bytes], # noqa: ARG002 + ) -> bytes: + return randomness_source(timestamp) + + +class LaxMonotonicPolicy: + """Lax monotonicity policy. + + Increments the randomness by 1 if generated within the same millisecond. + If the randomness overflows, it regenerates fresh randomness instead of raising + an error or sleeping. + """ + + def __init__(self) -> None: + self.prev_timestamp = constants.MIN_TIMESTAMP + self.prev_randomness = constants.MIN_RANDOMNESS + + def resolve_randomness( + self, + timestamp: int, + randomness_source: Callable[[int], bytes], + increment_bytes: Callable[[bytes], bytes], + ) -> bytes: + if timestamp == self.prev_timestamp: + if self.prev_randomness == constants.MAX_RANDOMNESS: + randomness = randomness_source(timestamp) + else: + randomness = increment_bytes(self.prev_randomness) + else: + randomness = randomness_source(timestamp) + + self.prev_randomness = randomness + self.prev_timestamp = timestamp + return randomness + + class ULIDGenerator: """Generator for creating universally unique lexicographically sortable identifiers (ULIDs). @@ -48,12 +146,12 @@ def __init__( self, clock: Callable[[], int] | None = None, randomness: Callable[[int], bytes] | None = None, + policy: MonotonicityPolicy | None = None, ) -> None: self.lock = Lock() self.clock = clock or self._default_clock self.randomness_source = randomness or self._default_randomness - self.prev_timestamp = constants.MIN_TIMESTAMP - self.prev_randomness = constants.MIN_RANDOMNESS + self.policy = policy or StrictMonotonicPolicy() @staticmethod def _default_clock() -> int: @@ -90,15 +188,11 @@ def generate(self, timestamp: float | datetime | None = None) -> ULID: ts_val = self.timestamp() with self.lock: - if ts_val == 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 = self.randomness_source(ts_val) - - self.prev_randomness = randomness - self.prev_timestamp = ts_val + randomness = self.policy.resolve_randomness( + ts_val, + self.randomness_source, + self.increment_bytes, + ) timestamp_bytes = int.to_bytes(ts_val, constants.TIMESTAMP_LEN, "big") return ULID.from_bytes(timestamp_bytes + randomness) From 5f05acef09a745a10c130ecefff6f9d7e454bb36 Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 12:49:09 +0200 Subject: [PATCH 4/8] feat(core): support direct policy injection in ULID value producers --- tests/test_ulid.py | 23 ++++++++++++++++++++- ulid/__init__.py | 51 +++++++++++++++++++++++++++++++++++++--------- 2 files changed, 63 insertions(+), 11 deletions(-) diff --git a/tests/test_ulid.py b/tests/test_ulid.py index ba13474..1967593 100644 --- a/tests/test_ulid.py +++ b/tests/test_ulid.py @@ -480,4 +480,25 @@ def test_lax_monotonic_policy() -> None: # Generating again should regenerate fresh randomness instead of raising ValueError or sleeping ulid3 = generator.generate() - assert ulid3 is not None + assert isinstance(ulid3, ULID) + + +def test_ulid_policy_injection() -> None: + # Ensure users can configure the policy via ULID constructors / creators directly + policy = PureRandomPolicy() + + # 1. Default constructor + ulid1 = ULID(policy=policy) + assert isinstance(ulid1, ULID) + + # 2. from_datetime + ulid2 = ULID.from_datetime(datetime.now(timezone.utc), policy=policy) + assert isinstance(ulid2, ULID) + + # 3. from_timestamp + ulid3 = ULID.from_timestamp(time.time(), policy=policy) + assert isinstance(ulid3, ULID) + + # 4. parse + ulid4 = ULID.parse(time.time(), policy=policy) + assert isinstance(ulid4, ULID) diff --git a/ulid/__init__.py b/ulid/__init__.py index 3eb866e..cf59cbe 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -205,6 +205,12 @@ def increment_bytes(self, value: bytes) -> bytes: _default_generator = ULIDGenerator() +def _resolve_generator(policy: MonotonicityPolicy | None = None) -> ULIDGenerator: + if policy is not None: + return ULIDGenerator(policy=policy) + return _default_generator + + @functools.total_ordering class ULID: """The :class:`ULID` object consists of a timestamp part of 48 bits and of 80 random bits. @@ -231,13 +237,27 @@ class ULID: 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, + *, + policy: MonotonicityPolicy | 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 _default_generator.generate().bytes + if value is None: + gen = _resolve_generator(policy) + self.bytes: bytes = gen.generate().bytes + else: + self.bytes = value @classmethod - def from_datetime(cls, value: datetime) -> Self: + def from_datetime( + cls, + value: datetime, + *, + policy: MonotonicityPolicy | 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. @@ -249,10 +269,15 @@ def from_datetime(cls, value: datetime) -> Self: """ if not isinstance(value, datetime): raise TypeError("Value has to be of type datetime") - return cls.from_timestamp(value.timestamp()) + return cls.from_timestamp(value.timestamp(), policy=policy) @classmethod - def from_timestamp(cls, value: float) -> Self: + def from_timestamp( + cls, + value: float, + *, + policy: MonotonicityPolicy | 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. @@ -265,7 +290,8 @@ def from_timestamp(cls, value: float) -> Self: """ if not isinstance(value, (int, float)): raise TypeError("Value has to be of type int or float") - return cls.from_bytes(_default_generator.generate(value).bytes) + gen = _resolve_generator(policy) + return cls.from_bytes(gen.generate(value).bytes) @classmethod def from_uuid(cls, value: uuid.UUID) -> Self: @@ -311,7 +337,12 @@ def from_int(cls, value: int) -> Self: return cls(int.to_bytes(value, constants.BYTES_LEN, "big")) @classmethod - def parse(cls, value: Any) -> Self: + def parse( + cls, + value: Any, + *, + policy: MonotonicityPolicy | None = None, + ) -> Self: """Create a new :class:`ULID`-object from a given value. .. note:: @@ -334,11 +365,11 @@ def parse(cls, value: Any) -> Self: if isinstance(value, int): if len(str(value)) == constants.INT_REPR_LEN: return cls.from_int(value) - return cls.from_timestamp(value) + return cls.from_timestamp(value, policy=policy) if isinstance(value, float): - return cls.from_timestamp(value) + return cls.from_timestamp(value, policy=policy) if isinstance(value, datetime): - return cls.from_datetime(value) + return cls.from_datetime(value, policy=policy) if isinstance(value, bytes): return cls.from_bytes(value) raise TypeError(f"Cannot parse ULID from type {type(value)}") From 07271a9e9a92941585bd8bbddd77dd67979675a5 Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 13:02:41 +0200 Subject: [PATCH 5/8] refactor(core): DRY up stateful monotonic policies with BaseMonotonicPolicy --- ulid/__init__.py | 73 +++++++++++++++++++++++++++++++----------------- 1 file changed, 47 insertions(+), 26 deletions(-) diff --git a/ulid/__init__.py b/ulid/__init__.py index cf59cbe..5fc0b7f 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -12,6 +12,12 @@ from typing import Protocol from typing import TYPE_CHECKING + +try: + from typing import override +except ImportError: + from typing_extensions import override # type: ignore + from ulid import base32 from ulid import constants @@ -60,12 +66,8 @@ def resolve_randomness( ... -class StrictMonotonicPolicy: - """Strict monotonicity policy. - - Always increments the randomness by 1 if generated within the same millisecond. - Raises ValueError on millisecond randomness exhaustion. - """ +class BaseMonotonicPolicy: + """Base class for stateful monotonic policies.""" def __init__(self) -> None: self.prev_timestamp = constants.MIN_TIMESTAMP @@ -79,8 +81,9 @@ def resolve_randomness( ) -> bytes: if timestamp == self.prev_timestamp: if self.prev_randomness == constants.MAX_RANDOMNESS: - raise ValueError("Randomness within same millisecond exhausted") - randomness = increment_bytes(self.prev_randomness) + randomness = self._on_overflow(timestamp, randomness_source) + else: + randomness = increment_bytes(self.prev_randomness) else: randomness = randomness_source(timestamp) @@ -88,6 +91,38 @@ def resolve_randomness( self.prev_timestamp = timestamp return randomness + def _on_overflow( + self, + timestamp: int, + randomness_source: Callable[[int], bytes], + ) -> bytes: + """Handle same-millisecond randomness exhaustion. + + Args: + timestamp (int): The current timestamp in milliseconds. + randomness_source (Callable[[int], bytes]): A callable to get fresh random bytes. + + Returns: + bytes: The resolved randomness bytes. + """ + raise NotImplementedError + + +class StrictMonotonicPolicy(BaseMonotonicPolicy): + """Strict monotonicity policy. + + Always increments the randomness by 1 if generated within the same millisecond. + Raises ValueError on millisecond randomness exhaustion. + """ + + @override + def _on_overflow( + self, + timestamp: int, + randomness_source: Callable[[int], bytes], + ) -> bytes: + raise ValueError("Randomness within same millisecond exhausted") + class PureRandomPolicy: """Pure random policy. @@ -104,7 +139,7 @@ def resolve_randomness( return randomness_source(timestamp) -class LaxMonotonicPolicy: +class LaxMonotonicPolicy(BaseMonotonicPolicy): """Lax monotonicity policy. Increments the randomness by 1 if generated within the same millisecond. @@ -112,27 +147,13 @@ class LaxMonotonicPolicy: an error or sleeping. """ - def __init__(self) -> None: - self.prev_timestamp = constants.MIN_TIMESTAMP - self.prev_randomness = constants.MIN_RANDOMNESS - - def resolve_randomness( + @override + def _on_overflow( self, timestamp: int, randomness_source: Callable[[int], bytes], - increment_bytes: Callable[[bytes], bytes], ) -> bytes: - if timestamp == self.prev_timestamp: - if self.prev_randomness == constants.MAX_RANDOMNESS: - randomness = randomness_source(timestamp) - else: - randomness = increment_bytes(self.prev_randomness) - else: - randomness = randomness_source(timestamp) - - self.prev_randomness = randomness - self.prev_timestamp = timestamp - return randomness + return randomness_source(timestamp) class ULIDGenerator: From 096fdd7fdde2c886fc3ab75c0ec963169525311d Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 13:07:02 +0200 Subject: [PATCH 6/8] feat(core): declare BaseMonotonicPolicy as an Abstract Base Class --- ulid/__init__.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/ulid/__init__.py b/ulid/__init__.py index 5fc0b7f..6c2265e 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -1,5 +1,6 @@ from __future__ import annotations +import abc import functools import os import time @@ -66,7 +67,7 @@ def resolve_randomness( ... -class BaseMonotonicPolicy: +class BaseMonotonicPolicy(abc.ABC): """Base class for stateful monotonic policies.""" def __init__(self) -> None: @@ -91,6 +92,7 @@ def resolve_randomness( self.prev_timestamp = timestamp return randomness + @abc.abstractmethod def _on_overflow( self, timestamp: int, From 97434132582ed419ebae5d24f30d9856878a1c65 Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 13:10:08 +0200 Subject: [PATCH 7/8] refactor(core): introduce RandomnessSource and RandomnessIncrementer type aliases --- ulid/__init__.py | 32 ++++++++++++++++++-------------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/ulid/__init__.py b/ulid/__init__.py index 6c2265e..9e5fc93 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -5,6 +5,7 @@ import os import time import uuid +from collections.abc import Callable from datetime import datetime from datetime import timezone from threading import Lock @@ -25,7 +26,6 @@ if TYPE_CHECKING: # pragma: no cover import sys - from collections.abc import Callable from pydantic import GetCoreSchemaHandler from pydantic import ValidatorFunctionWrapHandler @@ -45,21 +45,25 @@ __version__ = version("python-ulid") +RandomnessSource = Callable[[int], bytes] +RandomnessIncrementer = Callable[[bytes], bytes] + + class MonotonicityPolicy(Protocol): """Protocol defining the interface for monotonicity and randomness resolution policies.""" def resolve_randomness( self, timestamp: int, - randomness_source: Callable[[int], bytes], - increment_bytes: Callable[[bytes], bytes], + randomness_source: RandomnessSource, + increment_bytes: RandomnessIncrementer, ) -> bytes: """Resolve randomness for a given timestamp. Args: timestamp (int): The current timestamp in milliseconds. - randomness_source (Callable[[int], bytes]): A callable to get fresh random bytes. - increment_bytes (Callable[[bytes], bytes]): A callable to increment random bytes. + randomness_source (RandomnessSource): A callable to get fresh random bytes. + increment_bytes (RandomnessIncrementer): A callable to increment random bytes. Returns: bytes: The resolved randomness bytes (80 bits). @@ -77,8 +81,8 @@ def __init__(self) -> None: def resolve_randomness( self, timestamp: int, - randomness_source: Callable[[int], bytes], - increment_bytes: Callable[[bytes], bytes], + randomness_source: RandomnessSource, + increment_bytes: RandomnessIncrementer, ) -> bytes: if timestamp == self.prev_timestamp: if self.prev_randomness == constants.MAX_RANDOMNESS: @@ -96,13 +100,13 @@ def resolve_randomness( def _on_overflow( self, timestamp: int, - randomness_source: Callable[[int], bytes], + randomness_source: RandomnessSource, ) -> bytes: """Handle same-millisecond randomness exhaustion. Args: timestamp (int): The current timestamp in milliseconds. - randomness_source (Callable[[int], bytes]): A callable to get fresh random bytes. + randomness_source (RandomnessSource): A callable to get fresh random bytes. Returns: bytes: The resolved randomness bytes. @@ -121,7 +125,7 @@ class StrictMonotonicPolicy(BaseMonotonicPolicy): def _on_overflow( self, timestamp: int, - randomness_source: Callable[[int], bytes], + randomness_source: RandomnessSource, ) -> bytes: raise ValueError("Randomness within same millisecond exhausted") @@ -135,8 +139,8 @@ class PureRandomPolicy: def resolve_randomness( self, timestamp: int, - randomness_source: Callable[[int], bytes], - increment_bytes: Callable[[bytes], bytes], # noqa: ARG002 + randomness_source: RandomnessSource, + increment_bytes: RandomnessIncrementer, # noqa: ARG002 ) -> bytes: return randomness_source(timestamp) @@ -153,7 +157,7 @@ class LaxMonotonicPolicy(BaseMonotonicPolicy): def _on_overflow( self, timestamp: int, - randomness_source: Callable[[int], bytes], + randomness_source: RandomnessSource, ) -> bytes: return randomness_source(timestamp) @@ -168,7 +172,7 @@ class ULIDGenerator: def __init__( self, clock: Callable[[], int] | None = None, - randomness: Callable[[int], bytes] | None = None, + randomness: RandomnessSource | None = None, policy: MonotonicityPolicy | None = None, ) -> None: self.lock = Lock() From 00c8e18c1dd244b419fd883fd79c385712cf84d8 Mon Sep 17 00:00:00 2001 From: Martin Domke Date: Mon, 20 Jul 2026 16:15:14 +0200 Subject: [PATCH 8/8] feat(core): finalize public generator API and swappable default generator Consolidate ULID generation behind a single ULIDGenerator seam and expose it as the public surface for customizing generation: - Remove the per-call `policy` keyword from ULID(), from_timestamp, from_datetime and parse. The value object now delegates to a module-level, reassignable `default_generator`; callers wanting a custom policy build a ULIDGenerator and call generate(). - Collapse timestamp coercion into ULIDGenerator._normalize_timestamp so that generate() is the sole normalizer (datetime/int/float/None -> ms), with the narrow TypeError guards kept on the ULID.from_* constructors. - Narrow the MonotonicityPolicy interface: drop the increment_bytes parameter and move the increment into BaseMonotonicPolicy._increment. - Document ULIDGenerator, the monotonicity policies and default_generator in the API reference and add a "Generators and policies" section to the README. - Add 4.0.0 (breaking: ValueProvider/ULID.provider removed) and 3.2.1 changelog entries, and record the Keep a Changelog convention in AGENTS.md. --- AGENTS.md | 12 ++++++ CHANGELOG.rst | 40 +++++++++++++++++ README.rst | 58 +++++++++++++++++++++++++ docs/source/api.rst | 39 +++++++++++++++++ docs/source/index.rst | 4 ++ tests/test_ulid.py | 40 +++++++++-------- ulid/__init__.py | 99 +++++++++++++++++-------------------------- 7 files changed, 216 insertions(+), 76 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2b0a264..72928f3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -46,6 +46,18 @@ We require full, strict type annotations across the entire codebase. --- +## 📝 Changelog + +We maintain a human-readable changelog in [CHANGELOG.rst](file:///Users/martin.domke/Source/private/ulid/CHANGELOG.rst) following the **[Keep a Changelog](https://keepachangelog.com/en/1.1.0/)** conventions. + +- **Section headings**: Group entries under the standard headings, in this order: `Added`, `Changed`, `Deprecated`, `Removed`, `Fixed`, `Security`. Omit any section that has no entries. +- **Removals belong under `Removed`**: Do *not* fold removed or renamed public APIs into `Changed`. +- **Versioning**: Follow [Semantic Versioning](https://semver.org). A breaking change (e.g. a removed public API) requires a **major** version bump and a `.. warning::` admonition describing the migration path. +- **Entry format**: Each release has a ``` `X.Y.Z`_ - YYYY-MM-DD ``` heading plus a matching compare link at the bottom of the file (``.. _X.Y.Z: https://github.com/mdomke/python-ulid/compare/PREV...X.Y.Z``). +- **reStructuredText**: Keep lines within **100 characters** (enforced by `doc8`) and reference public symbols with Sphinx roles such as `:class:` and `:meth:` — the changelog is included into the rendered documentation. + +--- + ## 🛠️ Verification Workflow Before completing any task, you **MUST** run the verification commands to ensure no regressions or style issues are introduced. diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 1736d97..0a771d0 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -5,6 +5,44 @@ Changelog Versions follow `Semantic Versioning `_ +`4.0.0`_ - 2026-07-20 +--------------------- + +.. warning:: + + **Breaking change:** the ``ValueProvider`` class and the ``ULID.provider`` class attribute + have been removed. ULID generation is now handled by the new :class:`.ULIDGenerator` together + with pluggable monotonicity policies. To customize generation, construct a + :class:`.ULIDGenerator` (optionally with a custom clock, randomness source, or policy) and + either call its ``generate()`` method or assign it to ``ulid.default_generator``. + +Added +~~~~~ +* Added a public :class:`.ULIDGenerator` class that encapsulates ULID generation and can be + configured with a custom clock, randomness source, and monotonicity policy. +* Added pluggable monotonicity policies: :class:`.StrictMonotonicPolicy` (the default), + :class:`.LaxMonotonicPolicy` and :class:`.PureRandomPolicy`, together with the + :class:`.MonotonicityPolicy` protocol and the :class:`.BaseMonotonicPolicy` base class for + implementing custom policies. +* Added a module-level ``ulid.default_generator`` that can be reassigned to route ``ULID()`` and + the ``ULID.from_*`` constructors through a custom :class:`.ULIDGenerator`. + +Removed +~~~~~~~ +* Removed the ``ValueProvider`` class and the ``ULID.provider`` attribute in favour of + :class:`.ULIDGenerator` and the monotonicity policies. Code that replaced ``ULID.provider`` or + subclassed ``ValueProvider`` must migrate to a custom :class:`.ULIDGenerator` assigned to + ``ulid.default_generator``. +* Removed the internal ``validate_type`` decorator. The ``ULID.from_*`` constructors still raise + ``TypeError`` for arguments of the wrong type, so runtime behaviour is unchanged. + +`3.2.1`_ - 2026-07-17 +--------------------- +Fixed +~~~~~ +* Corrected the build and publish pipeline and the generated source distribution. This release + contains no changes to the library code. + `3.2.0`_ - 2026-07-17 --------------------- Added @@ -220,6 +258,8 @@ Changed * The package now has no external dependencies. * The test-coverage has been raised to 100%. +.. _4.0.0: https://github.com/mdomke/python-ulid/compare/3.2.1...4.0.0 +.. _3.2.1: https://github.com/mdomke/python-ulid/compare/3.2.0...3.2.1 .. _3.2.0: https://github.com/mdomke/python-ulid/compare/3.1.0...3.2.0 .. _3.1.0: https://github.com/mdomke/python-ulid/compare/3.0.0...3.1.0 .. _3.0.0: https://github.com/mdomke/python-ulid/compare/2.7.0...3.0.0 diff --git a/README.rst b/README.rst index 7115d57..b491d19 100644 --- a/README.rst +++ b/README.rst @@ -165,6 +165,64 @@ timestamp, then :math:`r_2 = r_1 + 1`. .. monotonic-end +.. generators-begin + +Generators and policies +----------------------- + +Every ``ULID`` is produced by a ``ULIDGenerator``, which samples a clock for the timestamp, +sources entropy for the randomness, and enforces a *monotonicity policy*. The bare ``ULID()`` +constructor and the ``ULID.from_*`` factory methods delegate to a shared module-level +``default_generator``. + +For most use cases the default is all you need. To customize generation — a different clock, a +custom entropy source, or another monotonicity policy — create your own ``ULIDGenerator`` and +call ``generate()`` + +.. code-block:: pycon + + >>> from ulid import ULIDGenerator + >>> generator = ULIDGenerator() + >>> generator.generate() + ULID(01HB0N8Q4RCE7YB1M2VZK9WX3T) + +Choosing a monotonicity policy +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +A policy decides how the randomness is resolved when multiple ULIDs are generated within the same +millisecond. Three policies are available: + +* ``StrictMonotonicPolicy`` *(default)* — increments the randomness by 1 on a same-millisecond + collision and raises ``ValueError`` if the randomness is exhausted. This is the behaviour + described under `Monotonic Support`_. +* ``LaxMonotonicPolicy`` — increments like the strict policy, but regenerates fresh randomness + instead of raising when the randomness is exhausted. +* ``PureRandomPolicy`` — ignores previous state and always draws fresh randomness, maximizing + entropy at the cost of same-millisecond sort order. + +.. code-block:: pycon + + >>> from ulid import ULIDGenerator, PureRandomPolicy + >>> generator = ULIDGenerator(policy=PureRandomPolicy()) + >>> generator.generate() + ULID(01HB0N9F3TA5KDQ6ZE0WYV7MRC) + +Overriding the default generator +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +To make ``ULID()`` and the ``ULID.from_*`` constructors use a custom generator globally, reassign +``ulid.default_generator`` + +.. code-block:: pycon + + >>> import ulid + >>> from ulid import ULID, ULIDGenerator, LaxMonotonicPolicy + >>> ulid.default_generator = ULIDGenerator(policy=LaxMonotonicPolicy()) + >>> ULID() # now generated with the lax policy + ULID(01HB0NB7X2M4C8VKQ0ZF5WD9RA) + +.. generators-end + .. cli-begin Command line interface diff --git a/docs/source/api.rst b/docs/source/api.rst index 7f9c574..ebc1141 100644 --- a/docs/source/api.rst +++ b/docs/source/api.rst @@ -11,3 +11,42 @@ ULID .. autoclass:: ULID :members: + + +Generators +---------- + +Every :class:`ULID` is produced by a :class:`ULIDGenerator`. The bare :class:`ULID` constructor +and the ``ULID.from_*`` factory methods delegate to a shared module-level +:data:`default_generator`. Create your own :class:`ULIDGenerator` to customize the clock, the +randomness source, or the :class:`MonotonicityPolicy`. + +.. autoclass:: ULIDGenerator + :members: + +.. autodata:: default_generator + :no-value: + + +Monotonicity policies +--------------------- + +A monotonicity policy decides how the randomness component is resolved when several ULIDs are +generated within the same millisecond. Pass an instance to :class:`ULIDGenerator`. Any object +satisfying the :class:`MonotonicityPolicy` protocol can be used; stateful policies can subclass +:class:`BaseMonotonicPolicy` and only implement the overflow behaviour. + +.. autoclass:: MonotonicityPolicy + :members: + +.. autoclass:: BaseMonotonicPolicy + :members: + +.. autoclass:: StrictMonotonicPolicy + :members: + +.. autoclass:: LaxMonotonicPolicy + :members: + +.. autoclass:: PureRandomPolicy + :members: diff --git a/docs/source/index.rst b/docs/source/index.rst index 5d57b49..d636486 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -24,6 +24,10 @@ Release v\ |release| (:ref:`What's new `) :start-after: monotonic-begin :end-before: monotonic-end +.. include:: ../../README.rst + :start-after: generators-begin + :end-before: generators-end + .. include:: ../../README.rst :start-after: cli-begin :end-before: cli-end diff --git a/tests/test_ulid.py b/tests/test_ulid.py index 1967593..9c9c3c5 100644 --- a/tests/test_ulid.py +++ b/tests/test_ulid.py @@ -12,6 +12,7 @@ from pydantic import BaseModel from pydantic import ValidationError +import ulid as ulid_pkg from ulid import base32 from ulid import constants from ulid import LaxMonotonicPolicy @@ -483,22 +484,27 @@ def test_lax_monotonic_policy() -> None: assert isinstance(ulid3, ULID) -def test_ulid_policy_injection() -> None: - # Ensure users can configure the policy via ULID constructors / creators directly - policy = PureRandomPolicy() +def test_default_generator_is_strict() -> None: + # The default generator is public and enforces strict monotonicity out of the box + assert isinstance(ulid_pkg.default_generator, ULIDGenerator) + assert isinstance(ulid_pkg.default_generator.policy, StrictMonotonicPolicy) - # 1. Default constructor - ulid1 = ULID(policy=policy) - assert isinstance(ulid1, ULID) - # 2. from_datetime - ulid2 = ULID.from_datetime(datetime.now(timezone.utc), policy=policy) - assert isinstance(ulid2, ULID) - - # 3. from_timestamp - ulid3 = ULID.from_timestamp(time.time(), policy=policy) - assert isinstance(ulid3, ULID) - - # 4. parse - ulid4 = ULID.parse(time.time(), policy=policy) - assert isinstance(ulid4, ULID) +def test_default_generator_swappable(monkeypatch: pytest.MonkeyPatch) -> None: + # Reassigning default_generator routes ULID() and the from_* constructors through it. + # monkeypatch restores the original default at teardown, keeping tests isolated. + custom_ts = 123456789 + custom_rand = b"1234567890" + custom = ULIDGenerator(clock=lambda: custom_ts, randomness=lambda _: custom_rand) + monkeypatch.setattr(ulid_pkg, "default_generator", custom) + + # ULID() draws both timestamp (via the clock) and randomness from the swapped generator + now_ulid = ULID() + assert now_ulid.milliseconds == custom_ts + assert now_ulid.bytes[constants.TIMESTAMP_LEN :] == custom_rand + + # Explicit-timestamp constructors keep their own timestamp but still draw randomness + # from the swapped generator (each distinct timestamp yields fresh randomness). + assert ULID.from_timestamp(1000).bytes[constants.TIMESTAMP_LEN :] == custom_rand + assert ULID.from_datetime(utcnow()).bytes[constants.TIMESTAMP_LEN :] == custom_rand + assert ULID.parse(2000).bytes[constants.TIMESTAMP_LEN :] == custom_rand diff --git a/ulid/__init__.py b/ulid/__init__.py index 9e5fc93..3a1f615 100644 --- a/ulid/__init__.py +++ b/ulid/__init__.py @@ -46,7 +46,6 @@ RandomnessSource = Callable[[int], bytes] -RandomnessIncrementer = Callable[[bytes], bytes] class MonotonicityPolicy(Protocol): @@ -56,14 +55,12 @@ def resolve_randomness( self, timestamp: int, randomness_source: RandomnessSource, - increment_bytes: RandomnessIncrementer, ) -> bytes: """Resolve randomness for a given timestamp. Args: timestamp (int): The current timestamp in milliseconds. randomness_source (RandomnessSource): A callable to get fresh random bytes. - increment_bytes (RandomnessIncrementer): A callable to increment random bytes. Returns: bytes: The resolved randomness bytes (80 bits). @@ -82,13 +79,12 @@ def resolve_randomness( self, timestamp: int, randomness_source: RandomnessSource, - increment_bytes: RandomnessIncrementer, ) -> bytes: if timestamp == self.prev_timestamp: if self.prev_randomness == constants.MAX_RANDOMNESS: randomness = self._on_overflow(timestamp, randomness_source) else: - randomness = increment_bytes(self.prev_randomness) + randomness = self._increment(self.prev_randomness) else: randomness = randomness_source(timestamp) @@ -96,6 +92,10 @@ def resolve_randomness( self.prev_timestamp = timestamp return randomness + @staticmethod + def _increment(value: bytes) -> bytes: + return (int.from_bytes(value, byteorder="big") + 1).to_bytes(len(value), byteorder="big") + @abc.abstractmethod def _on_overflow( self, @@ -140,7 +140,6 @@ def resolve_randomness( self, timestamp: int, randomness_source: RandomnessSource, - increment_bytes: RandomnessIncrementer, # noqa: ARG002 ) -> bytes: return randomness_source(timestamp) @@ -165,8 +164,18 @@ def _on_overflow( class ULIDGenerator: """Generator for creating universally unique lexicographically sortable identifiers (ULIDs). - This class handles stateful monotonic generation of ULIDs, ensuring that identifiers - generated within the same millisecond are monotonically increasing. + Samples a clock for the timestamp, sources entropy for the randomness, and enforces a + :class:`MonotonicityPolicy` so that identifiers generated within the same millisecond are + monotonically increasing. Generation is guarded by a lock and is safe to share across + threads. + + Args: + clock: A callable returning the current time in milliseconds. Defaults to the system + clock. + randomness: A callable that, given a timestamp, returns fresh random bytes for the + randomness component. Defaults to :func:`os.urandom`. + policy: The :class:`MonotonicityPolicy` used to resolve randomness on same-millisecond + collisions. Defaults to :class:`StrictMonotonicPolicy`. """ def __init__( @@ -188,9 +197,11 @@ def _default_clock() -> int: def _default_randomness(_timestamp: int) -> bytes: return os.urandom(constants.RANDOMNESS_LEN) - def timestamp(self, value: float | None = None) -> int: + def _normalize_timestamp(self, value: float | datetime | None = None) -> int: if value is None: value = self.clock() + elif isinstance(value, datetime): + value = int(value.timestamp() * constants.MILLISECS_IN_SECS) elif isinstance(value, float): value = int(value * constants.MILLISECS_IN_SECS) if value > constants.MAX_TIMESTAMP: @@ -206,36 +217,25 @@ def generate(self, timestamp: float | datetime | None = None) -> ULID: Returns: ULID: A generated ULID. """ - ts_val: int - if isinstance(timestamp, datetime): - ts_val = self.timestamp(timestamp.timestamp()) - elif isinstance(timestamp, (int, float)): - ts_val = self.timestamp(timestamp) - else: - ts_val = self.timestamp() + ts = self._normalize_timestamp(timestamp) with self.lock: randomness = self.policy.resolve_randomness( - ts_val, + ts, self.randomness_source, - self.increment_bytes, ) - timestamp_bytes = int.to_bytes(ts_val, constants.TIMESTAMP_LEN, "big") - return ULID.from_bytes(timestamp_bytes + randomness) - - def increment_bytes(self, value: bytes) -> bytes: - length = len(value) - return (int.from_bytes(value, byteorder="big") + 1).to_bytes(length, byteorder="big") + ts_bytes = int.to_bytes(ts, constants.TIMESTAMP_LEN, "big") + return ULID.from_bytes(ts_bytes + randomness) -_default_generator = ULIDGenerator() - - -def _resolve_generator(policy: MonotonicityPolicy | None = None) -> ULIDGenerator: - if policy is not None: - return ULIDGenerator(policy=policy) - return _default_generator +#: The module-level generator used by ``ULID()`` and the ``ULID.from_*`` constructors. +#: Reassign it to route the default constructors through a custom +#: :class:`ULIDGenerator` (e.g. a different clock, randomness source, or +#: :class:`MonotonicityPolicy`):: +#: +#: ulid.default_generator = ULIDGenerator(policy=LaxMonotonicPolicy()) +default_generator = ULIDGenerator() @functools.total_ordering @@ -267,24 +267,16 @@ class ULID: def __init__( self, value: bytes | None = None, - *, - policy: MonotonicityPolicy | None = None, ) -> None: if value is not None and len(value) != constants.BYTES_LEN: raise ValueError("ULID has to be exactly 16 bytes long.") if value is None: - gen = _resolve_generator(policy) - self.bytes: bytes = gen.generate().bytes + self.bytes: bytes = default_generator.generate().bytes else: self.bytes = value @classmethod - def from_datetime( - cls, - value: datetime, - *, - policy: MonotonicityPolicy | None = None, - ) -> Self: + def from_datetime(cls, value: datetime) -> 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. @@ -296,15 +288,10 @@ def from_datetime( """ if not isinstance(value, datetime): raise TypeError("Value has to be of type datetime") - return cls.from_timestamp(value.timestamp(), policy=policy) + return cls.from_bytes(default_generator.generate(value).bytes) @classmethod - def from_timestamp( - cls, - value: float, - *, - policy: MonotonicityPolicy | None = None, - ) -> Self: + def from_timestamp(cls, value: float) -> 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. @@ -317,8 +304,7 @@ def from_timestamp( """ if not isinstance(value, (int, float)): raise TypeError("Value has to be of type int or float") - gen = _resolve_generator(policy) - return cls.from_bytes(gen.generate(value).bytes) + return cls.from_bytes(default_generator.generate(value).bytes) @classmethod def from_uuid(cls, value: uuid.UUID) -> Self: @@ -364,12 +350,7 @@ def from_int(cls, value: int) -> Self: return cls(int.to_bytes(value, constants.BYTES_LEN, "big")) @classmethod - def parse( - cls, - value: Any, - *, - policy: MonotonicityPolicy | None = None, - ) -> Self: + def parse(cls, value: Any) -> Self: """Create a new :class:`ULID`-object from a given value. .. note:: @@ -392,11 +373,11 @@ def parse( if isinstance(value, int): if len(str(value)) == constants.INT_REPR_LEN: return cls.from_int(value) - return cls.from_timestamp(value, policy=policy) + return cls.from_timestamp(value) if isinstance(value, float): - return cls.from_timestamp(value, policy=policy) + return cls.from_timestamp(value) if isinstance(value, datetime): - return cls.from_datetime(value, policy=policy) + return cls.from_datetime(value) if isinstance(value, bytes): return cls.from_bytes(value) raise TypeError(f"Cannot parse ULID from type {type(value)}")