Coverage for faststream / _internal / utils / nuid.py: 87%

1# Copyright 2016-2018 The NATS Authors 

2# Licensed under the Apache License, Version 2.0 (the "License"); 

3# you may not use this file except in compliance with the License. 

4# You may obtain a copy of the License at 

5# 

6# http://www.apache.org/licenses/LICENSE-2.0 

7# 

8# Unless required by applicable law or agreed to in writing, software 

9# distributed under the License is distributed on an "AS IS" BASIS, 

10# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

11# See the License for the specific language governing permissions and 

12# limitations under the License. 

13 

14from __future__ import annotations (empty)

15 

16from random import Random (empty)

17from secrets import randbelow, token_bytes (empty)

18from sys import maxsize as max_int (empty)

19 

20DIGITS = b"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" (empty)

21BASE = 62 (empty)

22PREFIX_LENGTH = 12 (empty)

23SEQ_LENGTH = 10 (empty)

24MAX_SEQ = BASE**10 (empty)

25MIN_INC = 33 (empty)

26MAX_INC = 333 (empty)

27INC = MAX_INC - MIN_INC (empty)

28TOTAL_LENGTH = PREFIX_LENGTH + SEQ_LENGTH (empty)

29 

30 

31class NUID: (empty)

32 """NUID created is a utility to create a new id. 

33 

34 NUID is an implementation of the approach for fast generation 

35 of unique identifiers used for inboxes in NATS. 

36 """ 

37 

38 def __init__(self) -> None: (empty)

39 self._prand = Random(randbelow(max_int)) # nosec B311 # noqa: S311 (empty)

40 self._seq = self._prand.randint(0, MAX_SEQ) (empty)

41 self._inc = MIN_INC + self._prand.randint(BASE + 1, INC) (empty)

42 self._prefix = bytearray() (empty)

43 self.randomize_prefix() (empty)

44 

45 def next(self) -> bytearray: (empty)

46 """Next returns the next unique identifier.""" 

47 self._seq += self._inc (empty)

48 if self._seq >= MAX_SEQ: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true(empty)

49 self.randomize_prefix() 

50 self.reset_sequential() 

51 

52 l_seq = self._seq (empty)

53 prefix = self._prefix[:] (empty)

54 suffix = bytearray(SEQ_LENGTH) (empty)

55 for i in reversed(range(SEQ_LENGTH)): (empty)

56 suffix[i] = DIGITS[int(l_seq) % BASE] (empty)

57 l_seq //= BASE (empty)

58 

59 prefix.extend(suffix) (empty)

60 return prefix (empty)

61 

62 def randomize_prefix(self) -> None: (empty)

63 random_bytes = token_bytes(PREFIX_LENGTH) (empty)

64 self._prefix = bytearray(DIGITS[c % BASE] for c in random_bytes) (empty)

65 

66 def reset_sequential(self) -> None: (empty)

67 self._seq = self._prand.randint(0, MAX_SEQ) 

68 self._inc = MIN_INC + self._prand.randint(0, INC)