Publisher

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

@override
async def publish(  # type: ignore[override]
    self,
    *messages: SendableMessage,
    message: SendableMessage = "",
    key: Optional[bytes] = None,
    partition: Optional[int] = None,
    timestamp_ms: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None,
    correlation_id: Optional[str] = None,
) -> None:
    """Publish messages to a topic.

    Args:
        *messages: Variable length argument list of SendableMessage objects.
        message: A SendableMessage object. Default is an empty string.
        key: Optional bytes object representing the message key.
        partition: Optional integer representing the partition to publish the message to.
        timestamp_ms: Optional integer representing the timestamp of the message in milliseconds.
        headers: Optional dictionary of header key-value pairs.
        correlation_id: Optional string representing the correlation ID of the message.

    Returns:
        None

    Raises:
        AssertionError: If `_producer` is not set up.
        AssertionError: If `batch` flag is not set and there are multiple messages.
        ValueError: If `message` is not a sequence when `messages` is empty.

    """
    assert self._producer, NOT_CONNECTED_YET  # nosec B101
    assert (  # nosec B101
        self.batch or len(messages) < 2
    ), "You can't send multiple messages without `batch` flag"
    assert self.topic, "You have to specify outgoing topic"  # nosec B101

    if not self.batch:
        return await self._producer.publish(
            message=next(iter(messages), message),
            topic=self.topic,
            key=key or self.key,
            partition=partition or self.partition,
            timestamp_ms=timestamp_ms or self.timestamp_ms,
            correlation_id=correlation_id,
            headers=headers or self.headers,
            reply_to=self.reply_to or "",
        )
    else:
        to_send: Sequence[SendableMessage]
        if not messages:
            if not isinstance(message, Sequence):
                raise TypeError(
                    f"Message: {message} should be Sequence type to send in batch"
                )
            else:
                to_send = message
        else:
            to_send = messages

        await self._producer.publish_batch(
            *to_send,
            topic=self.topic,
            partition=partition or self.partition,
            timestamp_ms=timestamp_ms or self.timestamp_ms,
            headers=headers or self.headers,
        )
        return None

def reset_test(self) -> None:
    self._fake_handler = False
    self.mock = None

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(kafka=kafka.ChannelBinding(topic=self.topic)),
        )
    }

def set_test(
    self,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    self.mock = mock
    self._fake_handler = with_fake