Skip to content

NatsRoute

faststream.nats.broker.router.NatsRoute #

NatsRoute(
    call: Callable[..., SendableMessage]
    | Callable[..., Awaitable[SendableMessage]],
    subject: str,
    publishers: Iterable[NatsPublisher] = (),
    queue: str = "",
    pending_msgs_limit: int | None = None,
    pending_bytes_limit: int | None = None,
    max_msgs: int = 0,
    durable: str | None = None,
    config: Optional[ConsumerConfig] = None,
    ordered_consumer: bool = False,
    idle_heartbeat: float | None = None,
    flow_control: bool | None = None,
    deliver_policy: Optional[DeliverPolicy] = None,
    headers_only: bool | None = None,
    pull_sub: Optional[PullSub] = None,
    kv_watch: Union[str, KvWatch, None] = None,
    obj_watch: Union[bool, ObjWatch] = False,
    inbox_prefix: bytes = INBOX_PREFIX,
    stream: Union[str, JStream, None] = None,
    dependencies: Sequence[Dependant] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    max_workers: int | None = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    persistent: bool = True,
    codec: Optional[CodecProto] = None,
)

Bases: SubscriberRoute

Class to store delayed NatsBroker subscriber registration.

Initialized NatsRoute.

PARAMETER DESCRIPTION
call

Message handler function to wrap the same with @broker.subscriber(...) way.

TYPE: Callable[..., SendableMessage] | Callable[..., Awaitable[SendableMessage]]

subject

NATS subject to subscribe.

TYPE: str

publishers

Nats publishers to broadcast the handler result.

TYPE: Iterable[NatsPublisher] DEFAULT: ()

queue

Subscribers' NATS queue name. Subscribers with same queue name will be load balanced by the NATS server.

TYPE: str DEFAULT: ''

pending_msgs_limit

Limit of messages, considered by NATS server as possible to be delivered to the client without been answered.

TYPE: int | None DEFAULT: None

pending_bytes_limit

The number of bytes, considered by NATS server as possible to be delivered to the client without been answered.

TYPE: int | None DEFAULT: None

max_msgs

Consuming messages limiter. Automatically disconnect if reached.

TYPE: int DEFAULT: 0

durable

Name of the durable consumer to which the the subscription should be bound.

TYPE: str | None DEFAULT: None

config

Configuration of JetStream consumer to be subscribed with.

TYPE: Optional[ConsumerConfig] DEFAULT: None

ordered_consumer

Enable ordered consumer mode.

TYPE: bool DEFAULT: False

idle_heartbeat

Enable Heartbeats for a consumer to detect failures.

TYPE: float | None DEFAULT: None

flow_control

Enable Flow Control for a consumer.

TYPE: bool | None DEFAULT: None

deliver_policy

Deliver Policy to be used for subscription.

TYPE: Optional[DeliverPolicy] DEFAULT: None

headers_only

Should be message delivered without payload, only headers and metadata.

TYPE: bool | None DEFAULT: None

pull_sub

NATS Pull consumer parameters container. Should be used with stream only.

TYPE: Optional[PullSub] DEFAULT: None

kv_watch

KeyValue watch parameters container.

TYPE: Union[str, KvWatch, None] DEFAULT: None

obj_watch

ObjectStore watch parameters container.

TYPE: Union[bool, ObjWatch] DEFAULT: False

inbox_prefix

Prefix for generating unique inboxes, subjects with that prefix and NUID.

TYPE: bytes DEFAULT: INBOX_PREFIX

stream

Subscribe to NATS Stream with subject filter.

TYPE: Union[str, JStream, None] DEFAULT: None

dependencies

Dependencies list ([Dependant(),]) to apply to the subscriber.

TYPE: Sequence[Dependant] DEFAULT: ()

parser

Parser to map original nats-py Msg to FastStream one.

TYPE: Optional[CustomCallable] DEFAULT: None

decoder

Function to decode FastStream msg bytes body to python objects.

TYPE: Optional[CustomCallable] DEFAULT: None

max_workers

Number of workers to process messages concurrently.

TYPE: int | None DEFAULT: None

ack_policy

Acknowledgment policy for subscriber.

TYPE: AckPolicy DEFAULT: EMPTY

no_reply

Whether to disable FastStream RPC and Reply To auto responses or not.

TYPE: bool DEFAULT: False

title

AsyncAPI subscriber object title.

TYPE: str | None DEFAULT: None

description

AsyncAPI subscriber object description. "Uses decorated docstring as default.

TYPE: str | None DEFAULT: None

include_in_schema

Whetever to include operation in AsyncAPI schema or not.

TYPE: bool DEFAULT: True

persistent

Whether to make the subscriber persistent or not.

TYPE: bool DEFAULT: True

codec

Custom codec object.

TYPE: Optional[CodecProto] DEFAULT: None

Source code in faststream/nats/broker/router.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def __init__(
    self,
    call: Callable[..., "SendableMessage"]
    | Callable[..., Awaitable["SendableMessage"]],
    subject: str,
    publishers: Iterable[NatsPublisher] = (),
    queue: str = "",
    pending_msgs_limit: int | None = None,
    pending_bytes_limit: int | None = None,
    # Core arguments
    max_msgs: int = 0,
    # JS arguments
    durable: str | None = None,
    config: Optional["api.ConsumerConfig"] = None,
    ordered_consumer: bool = False,
    idle_heartbeat: float | None = None,
    flow_control: bool | None = None,
    deliver_policy: Optional["api.DeliverPolicy"] = None,
    headers_only: bool | None = None,
    # pull arguments
    pull_sub: Optional["PullSub"] = None,
    kv_watch: Union[str, "KvWatch", None] = None,
    obj_watch: Union[bool, "ObjWatch"] = False,
    inbox_prefix: bytes = api.INBOX_PREFIX,
    stream: Union[str, "JStream", None] = None,
    dependencies: Sequence["Dependant"] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    max_workers: int | None = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    persistent: bool = True,
    codec: Optional["CodecProto"] = None,
) -> None:
    """Initialized NatsRoute.

    Args:
        call:
            Message handler function to wrap the same with `@broker.subscriber(...)` way.
        subject:
            NATS subject to subscribe.
        publishers:
            Nats publishers to broadcast the handler result.
        queue:
            Subscribers' NATS queue name. Subscribers with same queue name will be load balanced by the NATS server.
        pending_msgs_limit:
            Limit of messages, considered by NATS server as possible to be delivered to the client without been answered.
        pending_bytes_limit:
            The number of bytes, considered by NATS server as possible to be delivered to the client without been answered.
        max_msgs:
            Consuming messages limiter. Automatically disconnect if reached.
        durable:
            Name of the durable consumer to which the the subscription should be bound.
        config:
            Configuration of JetStream consumer to be subscribed with.
        ordered_consumer:
            Enable ordered consumer mode.
        idle_heartbeat:
            Enable Heartbeats for a consumer to detect failures.
        flow_control:
            Enable Flow Control for a consumer.
        deliver_policy:
            Deliver Policy to be used for subscription.
        headers_only:
            Should be message delivered without payload, only headers and metadata.
        pull_sub:
            NATS Pull consumer parameters container. Should be used with `stream` only.
        kv_watch:
            KeyValue watch parameters container.
        obj_watch:
            ObjectStore watch parameters container.
        inbox_prefix:
            Prefix for generating unique inboxes, subjects with that prefix and NUID.
        stream:
            Subscribe to NATS Stream with `subject` filter.
        dependencies:
            Dependencies list (`[Dependant(),]`) to apply to the subscriber.
        parser:
            Parser to map original **nats-py** Msg to FastStream one.
        decoder:
            Function to decode FastStream msg bytes body to python objects.
        max_workers:
            Number of workers to process messages concurrently.
        ack_policy:
            Acknowledgment policy for subscriber.
        no_reply:
            Whether to disable **FastStream** RPC and Reply To auto responses or not.
        title:
            AsyncAPI subscriber object title.
        description:
            AsyncAPI subscriber object description.
            "Uses decorated docstring as default.
        include_in_schema:
            Whetever to include operation in AsyncAPI schema or not.
        persistent: Whether to make the subscriber persistent or not.
        codec: Custom codec object.
    """
    super().__init__(
        call,
        subject=subject,
        publishers=publishers,
        pending_msgs_limit=pending_msgs_limit,
        pending_bytes_limit=pending_bytes_limit,
        max_msgs=max_msgs,
        durable=durable,
        config=config,
        ordered_consumer=ordered_consumer,
        idle_heartbeat=idle_heartbeat,
        flow_control=flow_control,
        deliver_policy=deliver_policy,
        headers_only=headers_only,
        pull_sub=pull_sub,
        kv_watch=kv_watch,
        obj_watch=obj_watch,
        inbox_prefix=inbox_prefix,
        stream=stream,
        max_workers=max_workers,
        queue=queue,
        dependencies=dependencies,
        parser=parser,
        decoder=decoder,
        ack_policy=ack_policy,
        no_reply=no_reply,
        title=title,
        description=description,
        include_in_schema=include_in_schema,
        persistent=persistent,
        codec=codec,
    )

args instance-attribute #

args: Iterable[Any] = args

kwargs instance-attribute #

kwargs: dict[str, Any] = kwargs

call instance-attribute #

call: Callable[..., Any] = call

publishers instance-attribute #

publishers: Iterable[Any] = publishers