Skip to content

LogicPublisher

faststream.nats.publisher.LogicPublisher dataclass #

LogicPublisher(include_in_schema: bool = True, title: Optional[str] = None, _description: Optional[str] = None, _schema: Optional[Any] = None, _fake_handler: bool = False, subject: str = '', reply_to: str = '', headers: Optional[Dict[str, str]] = None, stream: Optional[JStream] = None, timeout: Optional[float] = None)

Bases: BasePublisher[Msg]

A class to represent a NATS publisher.

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)

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
143
144
145
def get_payloads(self) -> List[Tuple[AnyDict, str]]:
    payloads: List[Tuple[AnyDict, str]] = []

    if self._schema:
        params = {"response__": (self._schema, ...)}

        call_model: CallModel[Any, Any] = CallModel(
            call=lambda: None,
            model=create_model("Fake"),
            response_model=create_model(  # type: ignore[call-overload]
                "",
                __config__=get_config_base(),  # type: ignore[arg-type]
                **params,  # type: ignore[arg-type]
            ),
            params=params,
        )

        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

name #

name() -> str

Returns the name of the API operation.

Source code in faststream/asyncapi/base.py 
22
23
24
25
@abstractproperty
def name(self) -> str:
    """Returns the name of the API operation."""
    raise NotImplementedError()

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 
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
57
58
@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]

Returns the schema of the API operation as a dictionary of channel names and channel objects.

Source code in faststream/asyncapi/base.py 
27
28
29
def schema(self) -> Dict[str, Channel]:  # pragma: no cover
    """Returns the schema of the API operation as a dictionary of channel names and channel objects."""
    return {}

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