codekingpro/portable-devtools

1"""UUID helpers backed by uuid-utils."""2 3from __future__ import annotations4 5import time6import uuid7import warnings8from typing import Final9 10import xxhash11from uuid_utils.compat import uuid7 as _uuid_utils_uuid712 13_NANOS_PER_SECOND: Final = 1_000_000_00014 15 16def _to_timestamp_and_nanos(nanoseconds: int) -> tuple[int, int]:17    """Split a nanosecond timestamp into seconds and remaining nanoseconds."""18    seconds, nanos = divmod(nanoseconds, _NANOS_PER_SECOND)19    return seconds, nanos20 21 22def uuid7(nanoseconds: int | None = None) -> uuid.UUID:23    """Generate a UUID from a Unix timestamp in nanoseconds and random bits.24 25    UUIDv7 objects feature monotonicity within a millisecond.26 27    Args:28        nanoseconds: Optional ns timestamp. If not provided, uses current time.29    """30    # --- 48 ---   -- 4 --   --- 12 ---   -- 2 --   --- 30 ---   - 32 -31    # unix_ts_ms | version | counter_hi | variant | counter_lo | random32    #33    # 'counter = counter_hi | counter_lo' is a 42-bit counter constructed34    # with Method 1 of RFC 9562, §6.2, and its MSB is set to 0.35    #36    # 'random' is a 32-bit random value regenerated for every new UUID.37    #38    # If multiple UUIDs are generated within the same millisecond, the LSB39    # of 'counter' is incremented by 1. When overflowing, the timestamp is40    # advanced and the counter is reset to a random 42-bit integer with MSB41    # set to 0.42 43    # For now, just delegate to the uuid_utils implementation44    if nanoseconds is None:45        return _uuid_utils_uuid7()46    seconds, nanos = _to_timestamp_and_nanos(nanoseconds)47    return _uuid_utils_uuid7(timestamp=seconds, nanos=nanos)48 49 50def is_uuid_v7(uuid_obj: uuid.UUID) -> bool:51    """Check if a UUID is version 7.52 53    Args:54        uuid_obj: The UUID to check.55 56    Returns:57        True if the UUID is version 7, False otherwise.58    """59    return uuid_obj.version == 760 61 62_UUID_V7_WARNING_EMITTED = False63 64 65def warn_if_not_uuid_v7(uuid_obj: uuid.UUID, id_type: str) -> None:66    """Warn if a UUID is not version 7.67 68    Args:69        uuid_obj: The UUID to check.70        id_type: The type of ID (e.g., "run_id", "trace_id") for the warning message.71    """72    global _UUID_V7_WARNING_EMITTED73    if not is_uuid_v7(uuid_obj) and not _UUID_V7_WARNING_EMITTED:74        _UUID_V7_WARNING_EMITTED = True75        warnings.warn(76            (77                "LangSmith now uses UUID v7 for run and trace identifiers. "78                "This warning appears when passing custom IDs. "79                "Please use: from langsmith import uuid7\n"80                "            id = uuid7()\n"81                "Future versions will require UUID v7."82            ),83            UserWarning,84            stacklevel=3,85        )86 87 88def uuid7_deterministic(original_id: uuid.UUID, key: str) -> uuid.UUID:89    """Generate a deterministic UUID7 derived from an original UUID and a key.90 91    This function creates a new UUID that:92    - Preserves the timestamp from the original UUID if it's UUID v793    - Uses current time if the original is not UUID v794    - Uses deterministic bits derived from hashing the original + key with XXH3-12895    - Is valid UUID v7 format96 97    This is used for creating replica IDs that maintain time-ordering properties98    while being deterministic across distributed systems.99 100    Args:101        original_id: The source UUID (ideally UUID v7 to preserve timestamp).102        key: A string key used for deterministic derivation (e.g., project name).103 104    Returns:105        A new UUID v7 with preserved timestamp (if original is v7) and106        deterministic random bits.107 108    Example:109        >>> original = uuid7()110        >>> replica_id = uuid7_deterministic(original, "replica-project")111        >>> # Same inputs always produce same output112        >>> assert uuid7_deterministic(original, "replica-project") == replica_id113    """114    # Generate deterministic bytes from XXH3-128 hash of original + key115    hash_input = f"{original_id}:{key}".encode()116    h = xxhash.xxh3_128(hash_input).digest()117 118    # Build new UUID7:119    # UUID7 structure (RFC 9562):120    # [0-5]  48 bits: unix_ts_ms (timestamp in milliseconds)121    # [6]    4 bits: version (0111 = 7) + 4 bits rand_a122    # [7]    8 bits: rand_a (continued)123    # [8]    2 bits: variant (10) + 6 bits rand_b124    # [9-15] 56 bits: rand_b (continued)125 126    b = bytearray(16)127 128    # Check if original is UUID v7 - if so, preserve its timestamp129    # If not, use current time to ensure the derived UUID has a valid timestamp130    if is_uuid_v7(original_id):131        # Preserve timestamp from original UUID7 (bytes 0-5)132        b[0:6] = original_id.bytes[0:6]133    else:134        # Generate fresh timestamp for non-UUID7 inputs135        # This matches CPython 3.14's uuid7() implementation:136        # timestamp_ms = time.time_ns() // 1_000_000137        # Then convert to big-endian bytes138        timestamp_ms = time.time_ns() // 1_000_000139        # Mask to 48 bits and convert to big-endian bytes140        unix_ts_ms = timestamp_ms & 0xFFFF_FFFF_FFFF141        b[0:6] = unix_ts_ms.to_bytes(6, "big")142 143    # Set version 7 (0111) in high nibble + 4 bits from hash144    b[6] = 0x70 | (h[0] & 0x0F)145 146    # rand_a continued (8 bits from hash)147    b[7] = h[1]148 149    # Set variant (10) in high 2 bits + 6 bits from hash150    b[8] = 0x80 | (h[2] & 0x3F)151 152    # rand_b (56 bits = 7 bytes from hash)153    b[9:16] = h[3:10]154 155    return uuid.UUID(bytes=bytes(b))156