Skip to content

NUID

faststream.utils.nuid.NUID #

NUID()

NUID created is a utility to create a new id.

NUID is an implementation of the approach for fast generation of unique identifiers used for inboxes in NATS.

Source code in faststream/utils/nuid.py 
38
39
40
41
42
43
def __init__(self) -> None:
    self._prand = Random(randbelow(max_int))  # nosec B311
    self._seq = self._prand.randint(0, MAX_SEQ)
    self._inc = MIN_INC + self._prand.randint(BASE + 1, INC)
    self._prefix = bytearray()
    self.randomize_prefix()

next #

next()

Next returns the next unique identifier.

Source code in faststream/utils/nuid.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def next(self) -> bytearray:
    """Next returns the next unique identifier."""
    self._seq += self._inc
    if self._seq >= MAX_SEQ:
        self.randomize_prefix()
        self.reset_sequential()

    l_seq = self._seq
    prefix = self._prefix[:]
    suffix = bytearray(SEQ_LENGTH)
    for i in reversed(range(SEQ_LENGTH)):
        suffix[i] = DIGITS[int(l_seq) % BASE]
        l_seq //= BASE

    prefix.extend(suffix)
    return prefix

randomize_prefix #

randomize_prefix()
Source code in faststream/utils/nuid.py 
62
63
64
def randomize_prefix(self) -> None:
    random_bytes = token_bytes(PREFIX_LENGTH)
    self._prefix = bytearray(DIGITS[c % BASE] for c in random_bytes)

reset_sequential #

reset_sequential()
Source code in faststream/utils/nuid.py 
66
67
68
def reset_sequential(self) -> None:
    self._seq = self._prand.randint(0, MAX_SEQ)
    self._inc = MIN_INC + self._prand.randint(0, INC)