Skip to content

StreamBuilder

faststream.nats.helpers.StreamBuilder #

StreamBuilder()

A class to register stream-subjects pairs in Broker/Router.

Source code in faststream/nats/helpers/stream_builder.py 
11
12
13
14
15
def __init__(self) -> None:
    # stores stream: SubjectsCollection pairs
    # where SubjectsCollection contains subjects
    # made by current builder only
    self.objects: dict[str, tuple[JStream, SubjectsCollection]] = {}

objects instance-attribute #

objects = {}

create #

create(name)

Get an object.

Source code in faststream/nats/helpers/stream_builder.py 
22
23
24
25
26
27
28
29
def create(
    self,
    name: Union[str, "JStream", None],
) -> Optional["JStream"]:
    """Get an object."""
    if (stream := JStream.validate(name)) and (stream.name not in self.objects):
        self.objects[stream.name] = (stream, stream.subjects.copy())
    return stream

get #

get(stream, default=None)
Source code in faststream/nats/helpers/stream_builder.py 
31
32
33
34
35
36
37
38
def get(
    self,
    stream: Union["JStream", str, None],
    default: tuple["JStream", "SubjectsCollection"] | None = None,
) -> tuple["JStream", "SubjectsCollection"] | None:
    if stream := JStream.validate(stream):
        return self.objects.get(stream.name, default)
    return default

add_subject #

add_subject(stream, subject)
Source code in faststream/nats/helpers/stream_builder.py 
40
41
42
43
44
45
46
47
48
49
50
51
def add_subject(
    self,
    stream: Union["JStream", str, None],
    subject: str,
) -> None:
    if (stream := JStream.validate(stream)) and subject:
        stream, subjects = self.objects.get(
            stream.name,
            (stream, stream.subjects.copy()),
        )
        subjects.append(subject)
        self.objects[stream.name] = (stream, subjects)