Skip to content

RePublish

nats.js.api.RePublish dataclass #

RePublish(
    src: Optional[str] = None,
    dest: Optional[str] = None,
    headers_only: Optional[bool] = None,
)

Bases: Base

RePublish is for republishing messages once committed to a stream. The original subject cis remapped from the subject pattern to the destination pattern.

src class-attribute instance-attribute #

src: Optional[str] = None

dest class-attribute instance-attribute #

dest: Optional[str] = None

headers_only class-attribute instance-attribute #

headers_only: Optional[bool] = None

from_response classmethod #

from_response(resp: Dict[str, Any]) -> _B

Read the class instance from a server response.

Unknown fields are ignored ("open-world assumption").

Source code in nats/js/api.py 
145
146
147
148
149
150
151
152
153
154
155
@classmethod
def from_response(cls: type[_B], resp: Dict[str, Any]) -> _B:
    """Read the class instance from a server response.

    Unknown fields are ignored ("open-world assumption").
    """
    params = {}
    for field in fields(cls):
        if field.name in resp:
            params[field.name] = resp[field.name]
    return cls(**params)

evolve #

evolve(**params) -> _B

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

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

as_dict #

as_dict() -> Dict[str, object]

Return the object converted into an API-friendly dict.

Source code in nats/js/api.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def as_dict(self) -> Dict[str, object]:
    """Return the object converted into an API-friendly dict."""
    result = {}
    for field in fields(self):
        val = getattr(self, field.name)
        if val is None:
            continue
        if isinstance(val, Base):
            val = val.as_dict()
        if isinstance(val, list):
            if len(val) > 0 and isinstance(val[0], Base):
                val = [v.as_dict() for v in val if isinstance(v, Base)]
        result[field.name] = val
    return result