Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/uuid7.py: 90%

1from __future__ import annotations 1 ctx1g

2 

3import os 1 ctx1g

4import time 1 ctx1g

5import uuid 1 ctx1g

6from typing import Callable, Optional 1 ctx1g

7 

8 

9def _time_ms() -> int: 1 ctx1g

10 return time.time_ns() // 1_000_000 6 ctx1abcdef

11 

12 

13def uuid7( 1 ctx1g

14 ms: Optional[int] = None, 

15 time_func: Callable[[], int] = _time_ms, 

16) -> uuid.UUID: 

17 """ 

18 UUID v7, following the proposed extension to RFC4122 described in 

19 https://www.ietf.org/id/draft-peabody-dispatch-new-uuid-format-02.html. 

20 All representations (string, byte array, int) sort chronologically, 

21 with a potential time resolution of 50ns (if the system clock 

22 supports this). 

23 

24 Parameters 

25 ---------- 

26 

27 ms - Optional integer with the whole number of milliseconds 

28 since Unix epoch, to set the "as of" timestamp. 

29 

30 as_type - Optional string to return the UUID in a different format. 

31 A uuid.UUID (version 7, variant 0x10) is returned unless 

32 this is one of 'str', 'int', 'hex' or 'bytes'. 

33 

34 time_func - Set the time function, which must return integer 

35 milliseconds since the Unix epoch, midnight on 1-Jan-1970. 

36 Defaults to time.time_ns()/1e6. This is exposed because 

37 time.time_ns() may have a low resolution on Windows. 

38 

39 Returns 

40 ------- 

41 

42 A UUID object, or if as_type is specified, a string, int or 

43 bytes of length 16. 

44 

45 Implementation notes 

46 -------------------- 

47 

48 The 128 bits in the UUID are allocated as follows: 

49 - 36 bits of whole seconds 

50 - 24 bits of fractional seconds, giving approx 50ns resolution 

51 - 14 bits of sequential counter, if called repeatedly in same time tick 

52 - 48 bits of randomness 

53 plus, at locations defined by RFC4122, 4 bits for the 

54 uuid version (0b111) and 2 bits for the uuid variant (0b10). 

55 

56 0 1 2 3 

57 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 

58 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ 

59 | unix_ts_ms | 

60 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ 

61 | unix_ts_ms | ver | rand_a | 

62 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ 

63 |var| rand_b | 

64 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ 

65 | rand_b | 

66 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ 

67 

68 Indicative timings: 

69 - uuid.uuid4() 2.4us 

70 - uuid7() 3.7us 

71 - uuid7(as_type='int') 1.6us 

72 - uuid7(as_type='str') 2.5us 

73 

74 Examples 

75 -------- 

76 

77 ```python 

78 uuid7() 

79 # UUID('061cb26a-54b8-7a52-8000-2124e7041024') 

80 

81 for fmt in ('bytes', 'hex', 'int', 'str', 'uuid', None): 

82 print(fmt, repr(uuid7(as_type=fmt))) 

83 # bytes b'\x06\x1c\xb8\xfe\x0f\x0b|9\x80\x00\tjt\x85\xb3\xbb' 

84 # hex '061cb8fe0f0b7c3980011863b956b758' 

85 # int 8124504378724980906989670469352026642 

86 # str '061cb8fe-0f0b-7c39-8003-d44a7ee0bdf6' 

87 # uuid UUID('061cb8fe-0f0b-7c39-8004-0489578299f6') 

88 # None UUID('061cb8fe-0f0f-7df2-8000-afd57c2bf446') 

89 ``` 

90 """ 

91 if ms is None: 91 ↛ 94line 91 didn't jump to line 94 because the condition on line 91 was always true6 ctx1abcdef

92 ms = time_func() 6 ctx1abcdef

93 else: 

94 ms = int(ms) # Fail fast if not an int 

95 

96 rand_a = int.from_bytes(bytes=os.urandom(2), byteorder="big") 6 ctx1abcdef

97 rand_b = int.from_bytes(bytes=os.urandom(8), byteorder="big") 6 ctx1abcdef

98 uuid_bytes = uuidfromvalues(ms, rand_a, rand_b) 6 ctx1abcdef

99 

100 uuid_int = int.from_bytes(bytes=uuid_bytes, byteorder="big") 6 ctx1abcdef

101 return uuid.UUID(int=uuid_int) 6 ctx1abcdef

102 

103 

104def uuidfromvalues(unix_ts_ms: int, rand_a: int, rand_b: int): 1 ctx1g

105 version = 0x07 6 ctx1abcdef

106 var = 2 6 ctx1abcdef

107 rand_a &= 0xFFF 6 ctx1abcdef

108 rand_b &= 0x3FFFFFFFFFFFFFFF 6 ctx1abcdef

109 

110 final_bytes = unix_ts_ms.to_bytes(length=6, byteorder="big") 6 ctx1abcdef

111 final_bytes += ((version << 12) + rand_a).to_bytes(length=2, byteorder="big") 6 ctx1abcdef

112 final_bytes += ((var << 62) + rand_b).to_bytes(length=8, byteorder="big") 6 ctx1abcdef

113 

114 return final_bytes 6 ctx1abcdef

115 

116 

117def format_byte_array_as_uuid(arr: bytes): 1 ctx1g

118 return f"{arr[:4].hex()}-{arr[4:6].hex()}-{arr[6:8].hex()}-{arr[8:10].hex()}-{arr[10:].hex()}" 

119 

120 

121__all__ = ("uuid7",) 1 ctx1g