Skip to content

ABCPublisher

faststream.rabbit.shared.publisher.ABCPublisher dataclass #

ABCPublisher(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: ABC, BasePublisher[MsgType], BaseRMQInformation

A class representing an ABCPublisher.

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_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 
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 abstractmethod async #

publish(message: SendableMessage, correlation_id: Optional[str] = None, **kwargs: Any) -> Optional[SendableMessage]

Publish a message.

PARAMETER DESCRIPTION
message

The message to be published.

TYPE: SendableMessage

correlation_id

Optional correlation ID for the message.

TYPE: Optional[str] DEFAULT: None

**kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
Optional[SendableMessage]

The published message.
RAISES DESCRIPTION
NotImplementedError

If the method is not implemented.
Source code in faststream/broker/publisher.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@abstractmethod
async def publish(
    self,
    message: SendableMessage,
    correlation_id: Optional[str] = None,
    **kwargs: Any,
) -> Optional[SendableMessage]:
    """Publish a message.

    Args:
        message: The message to be published.
        correlation_id: Optional correlation ID for the message.
        **kwargs: Additional keyword arguments.

    Returns:
        The published message.

    Raises:
        NotImplementedError: If the method is not implemented.

    """
    raise NotImplementedError()

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