Coverage for langsmith/_internal/_uuid.py: 10%

1"""UUID helpers backed by uuid-utils.""" 

2 

3from __future__ import annotations 

4 

5import uuid 

6import warnings 

7from typing import Final 

8 

9from uuid_utils.compat import uuid7 as _uuid_utils_uuid7 

10 

11_NANOS_PER_SECOND: Final = 1_000_000_000 

12 

13 

14def _to_timestamp_and_nanos(nanoseconds: int) -> tuple[int, int]: 

15 """Split a nanosecond timestamp into seconds and remaining nanoseconds.""" 

16 seconds, nanos = divmod(nanoseconds, _NANOS_PER_SECOND) 

17 return seconds, nanos 

18 

19 

20def uuid7(nanoseconds: int | None = None) -> uuid.UUID: 

21 """Generate a UUID from a Unix timestamp in nanoseconds and random bits. 

22 

23 UUIDv7 objects feature monotonicity within a millisecond. 

24 

25 Args: 

26 nanoseconds: Optional ns timestamp. If not provided, uses current time. 

27 """ 

28 # --- 48 --- -- 4 -- --- 12 --- -- 2 -- --- 30 --- - 32 - 

29 # unix_ts_ms | version | counter_hi | variant | counter_lo | random 

30 # 

31 # 'counter = counter_hi | counter_lo' is a 42-bit counter constructed 

32 # with Method 1 of RFC 9562, §6.2, and its MSB is set to 0. 

33 # 

34 # 'random' is a 32-bit random value regenerated for every new UUID. 

35 # 

36 # If multiple UUIDs are generated within the same millisecond, the LSB 

37 # of 'counter' is incremented by 1. When overflowing, the timestamp is 

38 # advanced and the counter is reset to a random 42-bit integer with MSB 

39 # set to 0. 

40 

41 # For now, just delegate to the uuid_utils implementation 

42 if nanoseconds is None: 

43 return _uuid_utils_uuid7() 

44 seconds, nanos = _to_timestamp_and_nanos(nanoseconds) 

45 return _uuid_utils_uuid7(timestamp=seconds, nanos=nanos) 

46 

47 

48def is_uuid_v7(uuid_obj: uuid.UUID) -> bool: 

49 """Check if a UUID is version 7. 

50 

51 Args: 

52 uuid_obj: The UUID to check. 

53 

54 Returns: 

55 True if the UUID is version 7, False otherwise. 

56 """ 

57 return uuid_obj.version == 7 

58 

59 

60_UUID_V7_WARNING_EMITTED = False 

61 

62 

63def warn_if_not_uuid_v7(uuid_obj: uuid.UUID, id_type: str) -> None: 

64 """Warn if a UUID is not version 7. 

65 

66 Args: 

67 uuid_obj: The UUID to check. 

68 id_type: The type of ID (e.g., "run_id", "trace_id") for the warning message. 

69 """ 

70 global _UUID_V7_WARNING_EMITTED 

71 if not is_uuid_v7(uuid_obj) and not _UUID_V7_WARNING_EMITTED: 

72 _UUID_V7_WARNING_EMITTED = True 

73 warnings.warn( 

74 ( 

75 "LangSmith now uses UUID v7 for run and trace identifiers. " 

76 "This warning appears when passing custom IDs. " 

77 "Please use: from langsmith import uuid7\n" 

78 " id = uuid7()\n" 

79 "Future versions will require UUID v7." 

80 ), 

81 UserWarning, 

82 stacklevel=3, 

83 )