Skip to content

StreamSource

nats.js.api.StreamSource dataclass #

StreamSource(
    name: str,
    opt_start_seq: Optional[int] = None,
    filter_subject: Optional[str] = None,
    external: Optional[ExternalStream] = None,
    subject_transforms: Optional[
        List[SubjectTransform]
    ] = None,
)

Bases: Base

name instance-attribute #

name: str

opt_start_seq class-attribute instance-attribute #

opt_start_seq: Optional[int] = None

filter_subject class-attribute instance-attribute #

filter_subject: Optional[str] = None

external class-attribute instance-attribute #

external: Optional[ExternalStream] = None

subject_transforms class-attribute instance-attribute #

subject_transforms: Optional[List[SubjectTransform]] = None

from_response classmethod #

from_response(resp: Dict[str, Any])
Source code in nats/js/api.py 
160
161
162
163
164
@classmethod
def from_response(cls, resp: Dict[str, Any]):
    cls._convert(resp, "external", ExternalStream)
    cls._convert(resp, "subject_transforms", SubjectTransform)
    return super().from_response(resp)

as_dict #

as_dict() -> Dict[str, object]
Source code in nats/js/api.py 
166
167
168
169
170
171
172
def as_dict(self) -> Dict[str, object]:
    result = super().as_dict()
    if self.subject_transforms:
        result["subject_transforms"] = [
            tr.as_dict() for tr in self.subject_transforms
        ]
    return result

evolve #

evolve(**params) -> _B

Return a copy of the instance with the passed values replaced.

Source code in nats/js/api.py 
100
101
102
def evolve(self: _B, **params) -> _B:
    """Return a copy of the instance with the passed values replaced."""
    return replace(self, **params)