Skip to content

Publisher

faststream.nats.asyncapi.Publisher #

Bases: LogicPublisher

calls class-attribute instance-attribute #

calls: List[Callable[..., Any]] = field(
    init=False, default_factory=list, repr=False
)

description property #

description: Optional[str]

headers class-attribute instance-attribute #

headers: Optional[Dict[str, str]] = field(default=None)

include_in_schema class-attribute instance-attribute #

include_in_schema: bool = field(default=True)

mock class-attribute instance-attribute #

mock: Optional[MagicMock] = field(
    init=False, default=None, repr=False
)

name property #

name: str

reply_to class-attribute instance-attribute #

reply_to: str = field(default='')

stream class-attribute instance-attribute #

stream: Optional[JStream] = field(default=None)

subject class-attribute instance-attribute #

subject: str = field(default='')

timeout class-attribute instance-attribute #

timeout: Optional[float] = field(default=None)

title class-attribute instance-attribute #

title: Optional[str] = field(default=None)

get_payloads #

get_payloads() -> List[Tuple[AnyDict, str]]
Source code in faststream/broker/publisher.py 
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
def get_payloads(self) -> List[Tuple[AnyDict, str]]:
    payloads: List[Tuple[AnyDict, str]] = []

    if self._schema:
        call_model: CallModel[Any, Any] = CallModel(
            call=lambda: None,
            model=create_model("Fake"),
            response_model=create_model(
                "",
                __config__=get_config_base(),  # type: ignore[arg-type]
                response__=(self._schema, ...),
            ),
        )

        body = get_response_schema(
            call_model,
            prefix=f"{self.name}:Message",
        )
        if body:  # pragma: no branch
            payloads.append((body, ""))

    else:
        for call in self.calls:
            call_model = build_call_model(call)
            body = get_response_schema(
                call_model,
                prefix=f"{self.name}:Message",
            )
            if body:
                payloads.append((body, to_camelcase(unwrap(call).__name__)))

    return payloads

publish async #

publish(
    message: SendableMessage = "",
    reply_to: str = "",
    correlation_id: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    **producer_kwargs: Any
) -> Optional[DecodedMessage]
Source code in faststream/nats/publisher.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@override
async def publish(  # type: ignore[override]
    self,
    message: SendableMessage = "",
    reply_to: str = "",
    correlation_id: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    **producer_kwargs: Any,
) -> Optional[DecodedMessage]:
    assert self._producer, NOT_CONNECTED_YET  # nosec B101
    assert self.subject, "You have to specify outgoing subject"  # nosec B101

    extra: AnyDict = {
        "reply_to": reply_to or self.reply_to,
    }
    if self.stream is not None:
        extra.update(
            {
                "stream": self.stream.name,
                "timeout": self.timeout,
            }
        )

    return await self._producer.publish(
        message=message,
        subject=self.subject,
        headers=headers or self.headers,
        correlation_id=correlation_id,
        **extra,
        **producer_kwargs,
    )

reset_test #

reset_test() -> None
Source code in faststream/broker/publisher.py 
61
62
63
def reset_test(self) -> None:
    self._fake_handler = False
    self.mock = None

schema #

schema() -> Dict[str, Channel]
Source code in faststream/nats/asyncapi.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def schema(self) -> Dict[str, Channel]:
    if not self.include_in_schema:
        return {}

    payloads = self.get_payloads()

    return {
        self.name: Channel(
            description=self.description,
            publish=Operation(
                message=Message(
                    title=f"{self.name}:Message",
                    payload=resolve_payloads(payloads, "Publisher"),
                    correlationId=CorrelationId(
                        location="$message.header#/correlation_id"
                    ),
                ),
            ),
            bindings=ChannelBinding(
                nats=nats.ChannelBinding(
                    subject=self.subject,
                )
            ),
        )
    }

set_test #

set_test(mock: MagicMock, with_fake: bool) -> None
Source code in faststream/broker/publisher.py 
53
54
55
56
57
58
59
def set_test(
    self,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    self.mock = mock
    self._fake_handler = with_fake