Skip to content

StreamBuilder

faststream.nats.helpers.StreamBuilder #

StreamBuilder()

A class to build streams.

Initialize the builder.

Source code in faststream/nats/helpers/object_builder.py 
13
14
15
def __init__(self) -> None:
    """Initialize the builder."""
    self.objects = {}

objects instance-attribute #

objects = {}

create #

create(name)

Get an object.

Source code in faststream/nats/helpers/object_builder.py 
17
18
19
20
21
22
23
24
25
26
27
def create(
    self,
    name: Union[str, "JStream", None],
) -> Optional["JStream"]:
    """Get an object."""
    stream = JStream.validate(name)

    if stream is not None:
        stream = self.objects[stream.name] = self.objects.get(stream.name, stream)

    return stream