Method publisher

faststream/nats/fastapi/fastapi.py:990–1031

Creates long-living and AsyncAPI-documented publisher object. Args: subject: NATS subject to send message. headers: Message headers to store metainformation. **content-type** and **correlation_id** will be set automatically by framework anyway.

(  # type: ignore[override]
        self,
        subject: str,
        headers: dict[str, str] | None = None,
        reply_to: str = "",
        # JS
        stream: Union[str, "JStream", None] = None,
        timeout: float | None = None,
        # AsyncAPI information
        title: str | None = None,
        description: str | None = None,
        schema: Any | None = None,
        include_in_schema: bool = True,
    )

    @override
    def publisher(  # type: ignore[override]
        self,
        subject: str,
        headers: dict[str, str] | None = None,
        reply_to: str = "",
        # JS
        stream: Union[str, "JStream", None] = None,
        timeout: float | None = None,
        # AsyncAPI information
        title: str | None = None,
        description: str | None = None,
        schema: Any | None = None,
        include_in_schema: bool = True,
    ) -> "LogicPublisher":
        """Creates long-living and AsyncAPI-documented publisher object.

        Args:
            subject: NATS subject to send message.
            headers: Message headers to store metainformation.
                **content-type** and **correlation_id** will be set automatically by framework anyway.
                Can be overridden by `publish.headers` if specified.
            reply_to: NATS subject name to send response.
            stream: This option validates that the target `subject` is in presented stream.
                Can be omitted without any effect.
            timeout: Timeout to send message to NATS.
            title: AsyncAPI publisher object title.
            description: AsyncAPI publisher object description.
            schema: AsyncAPI publishing message type.
                Should be any python-native object annotation or `pydantic.BaseModel`.
            include_in_schema: Whetever to include operation in AsyncAPI schema or not.
        """
        return self.broker.publisher(
            subject,
            headers=headers,
            reply_to=reply_to,
            stream=stream,
            timeout=timeout,
            title=title,
            description=description,
            schema=schema,
            include_in_schema=include_in_schema,
        )