Skip to content

JStream

faststream.nats.js_stream.JStream #

JStream(
    name: str,
    *args: Any,
    declare: bool = True,
    **kwargs: Any
)

Bases: NameRequired

A class to represent a JetStream stream.

Initialize the JetStream stream.

PARAMETER DESCRIPTION
name

The stream name.

TYPE: str

*args

The arguments.

TYPE: Any DEFAULT: ()

declare

Whether to declare the stream.

TYPE: bool DEFAULT: True

**kwargs

The keyword arguments.

TYPE: Any DEFAULT: {}

Source code in faststream/nats/js_stream.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    name: str,
    *args: Any,
    declare: bool = True,
    **kwargs: Any,
) -> None:
    """Initialize the JetStream stream.

    Args:
        name: The stream name.
        *args: The arguments.
        declare: Whether to declare the stream.
        **kwargs: The keyword arguments.
    """
    super().__init__(
        name=name,
        declare=declare,
        subjects=[],
        config=StreamConfig(
            *args,
            name=name,
            **kwargs,  # type: ignore[misc]
        ),
    )

config instance-attribute #

config: StreamConfig

declare class-attribute instance-attribute #

declare: bool = Field(default=True)

name class-attribute instance-attribute #

name: str = Field(...)

subjects class-attribute instance-attribute #

subjects: list[str] = Field(default_factory=list)

validate classmethod #

validate(
    value: Union[str, NameRequiredCls, None], **kwargs: Any
) -> Optional[NameRequiredCls]

Validates a value.

PARAMETER DESCRIPTION
value

The value to be validated.

TYPE: Union[str, NameRequiredCls, None]

**kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
Optional[NameRequiredCls]

The validated value.
Source code in faststream/broker/schemas.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@classmethod
def validate(
    cls: Type[NameRequiredCls],
    value: Union[str, NameRequiredCls, None],
    **kwargs: Any,
) -> Optional[NameRequiredCls]:
    """Validates a value.

    Args:
        value: The value to be validated.
        **kwargs: Additional keyword arguments.

    Returns:
        The validated value.

    """
    if value is not None and isinstance(value, str):
        value = cls(value, **kwargs)
    return value