Skip to content

StreamBuilder

faststream.nats.helpers.StreamBuilder #

StreamBuilder()
Source code in faststream/nats/helpers.py 
 9
10
def __init__(self) -> None:
    self.streams = {}

streams instance-attribute #

streams: Dict[str, JStream] = {}

stream #

stream(
    name: Union[str, JStream, None],
    *args: Any,
    declare: bool = True,
    **kwargs: Any
) -> Optional[JStream]
Source code in faststream/nats/helpers.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
def stream(
    self,
    name: Union[str, JStream, None],
    *args: Any,
    declare: bool = True,
    **kwargs: Any,
) -> Optional[JStream]:
    stream = JStream.validate(name)

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

    return stream
Last update: 2023-11-13