Skip to content

LogicPublisher

faststream.rabbit.publisher.LogicPublisher dataclass #

LogicPublisher(queue: RabbitQueue = RabbitQueue(''), exchange: Optional[RabbitExchange] = None, _description: Optional[str] = None, virtual_host: str = '/', include_in_schema: bool = True, title: Optional[str] = None, _schema: Optional[Any] = None, _fake_handler: bool = False, routing_key: str = '', mandatory: bool = True, immediate: bool = False, persist: bool = False, timeout: TimeoutType = None, reply_to: Optional[str] = None, priority: Optional[int] = None, message_kwargs: AnyDict = dict())

Bases: ABCPublisher[IncomingMessage]

A class to publish messages for logic processing.

METHOD DESCRIPTION
publish

Publishes a message for logic processing.

calls class-attribute instance-attribute #

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

description property #

description: Optional[str]

exchange class-attribute instance-attribute #

exchange: Optional[RabbitExchange] = field(default=None)

immediate class-attribute instance-attribute #

immediate: bool = False

include_in_schema class-attribute instance-attribute #

include_in_schema: bool = field(default=True)

mandatory class-attribute instance-attribute #

mandatory: bool = True

message_kwargs class-attribute instance-attribute #

message_kwargs: AnyDict = field(default_factory=dict)

mock class-attribute instance-attribute #

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

persist class-attribute instance-attribute #

persist: bool = False

priority class-attribute instance-attribute #

priority: Optional[int] = None

queue class-attribute instance-attribute #

queue: RabbitQueue = field(default=RabbitQueue(''))

reply_to class-attribute instance-attribute #

reply_to: Optional[str] = None

routing property #

routing: str | None

routing_key class-attribute instance-attribute #

routing_key: str = ''

timeout class-attribute instance-attribute #

timeout: TimeoutType = None

title class-attribute instance-attribute #

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

virtual_host class-attribute instance-attribute #

virtual_host: str = '/'

get_payloads #

get_payloads() -> List[Tuple[AnyDict, str]]
Source code in faststream/broker/publisher.py
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
Source code in faststream/rabbit/publisher.py
    _producer : An optional AioPikaFastProducer object.

Methods:

publish async #

publish(message: AioPikaSendableMessage = '', *, rpc: bool = False, rpc_timeout: float | None = 30.0, raise_timeout: bool = False, correlation_id: str | None = None, priority: int | None = None, **message_kwargs: Any) -> ConfirmationFrameType | SendableMessage

Publish a message.

PARAMETER DESCRIPTION
message

The message to be published.

TYPE: AioPikaSendableMessage DEFAULT: ''

rpc

Whether the message is for RPC (Remote Procedure Call).

TYPE: bool DEFAULT: False

rpc_timeout

Timeout for RPC.

TYPE: float | None DEFAULT: 30.0

raise_timeout

Whether to raise an exception if timeout occurs.

TYPE: bool DEFAULT: False

correlation_id

Correlation ID for the message.

TYPE: str | None DEFAULT: None

priority

Priority for the message.

TYPE: int | None DEFAULT: None

**message_kwargs

Additional keyword arguments for the message.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
ConfirmationFrameType | SendableMessage

ConfirmationFrameType or SendableMessage: The result of the publish operation.

RAISES DESCRIPTION
AssertionError

If _producer is not set up.

Source code in faststream/rabbit/publisher.py
@override
async def publish(  # type: ignore[override]
    self,
    message: AioPikaSendableMessage = "",
    *,
    rpc: bool = False,
    rpc_timeout: Optional[float] = 30.0,
    raise_timeout: bool = False,
    correlation_id: Optional[str] = None,
    priority: Optional[int] = None,
    **message_kwargs: Any,
) -> Union[aiormq.abc.ConfirmationFrameType, SendableMessage]:
    """Publish a message.

    Args:
        message: The message to be published.
        rpc: Whether the message is for RPC (Remote Procedure Call).
        rpc_timeout: Timeout for RPC.
        raise_timeout: Whether to raise an exception if timeout occurs.
        correlation_id: Correlation ID for the message.
        priority: Priority for the message.
        **message_kwargs: Additional keyword arguments for the message.

    Returns:
        ConfirmationFrameType or SendableMessage: The result of the publish operation.

    Raises:
        AssertionError: If `_producer` is not set up.

    """
    assert self._producer, NOT_CONNECTED_YET  # nosec B101
    return await self._producer.publish(
        message=message,
        exchange=self.exchange,
        routing_key=self.routing,
        mandatory=self.mandatory,
        immediate=self.immediate,
        timeout=self.timeout,
        rpc=rpc,
        rpc_timeout=rpc_timeout,
        raise_timeout=raise_timeout,
        persist=self.persist,
        reply_to=self.reply_to,
        correlation_id=correlation_id,
        priority=priority or self.priority,
        **self.message_kwargs,
        **message_kwargs,
    )

reset_test #

reset_test() -> None
Source code in faststream/broker/publisher.py
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
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
def set_test(
    self,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    self.mock = mock
    self._fake_handler = with_fake