Function write

python/pathway/io/rabbitmq/__init__.py:252–387

Writes data into the specified RabbitMQ stream. The produced messages consist of the payload, corresponding to the values of the table that are serialized according to the chosen format. Two AMQP 1.0 application properties are always added: ``pathway_time`` (processing time) and ``pathw

(
    table: Table,
    uri: str,
    stream_name: str | ColumnReference,
    *,
    format: Literal["json", "plaintext", "raw"] = "json",
    value: ColumnReference | None = None,
    headers: Iterable[ColumnReference] | None = None,
    name: str | None = None,
    sort_by: Iterable[ColumnReference] | None = None,
    tls_settings: TLSSettings | None = None,
)

@check_arg_types
@trace_user_frame
def write(
    table: Table,
    uri: str,
    stream_name: str | ColumnReference,
    *,
    format: Literal["json", "plaintext", "raw"] = "json",
    value: ColumnReference | None = None,
    headers: Iterable[ColumnReference] | None = None,
    name: str | None = None,
    sort_by: Iterable[ColumnReference] | None = None,
    tls_settings: TLSSettings | None = None,
) -> None:
    """Writes data into the specified RabbitMQ stream.

    The produced messages consist of the payload, corresponding to the values of the
    table that are serialized according to the chosen format. Two AMQP 1.0 application
    properties are always added: ``pathway_time`` (processing time) and ``pathway_diff``
    (either ``1`` or ``-1``). If ``headers`` parameter is used, additional properties
    can be added to the message.

    There are several serialization formats supported: ``"json"``,
    ``"plaintext"`` and ``"raw"``.

    If the selected format is either ``"plaintext"`` or ``"raw"``, you also need to
    specify which column of the table corresponds to the payload of the produced message.
    It can be done by providing the ``value`` parameter.

    Args:
        table: The table for output.
        uri: The URI of the RabbitMQ server with Streams enabled, e.g.
            ``"rabbitmq-stream://guest:guest@localhost:5552"``.
        stream_name: The RabbitMQ stream where data will be written. The stream must
            already exist on the server. Can be a column reference for dynamic
            routing — each row will be written to the stream named by that column's
            value. All target streams must be pre-created.
        format: Format in which the data is put into RabbitMQ. Currently ``"json"``,
            ``"plaintext"`` and ``"raw"`` are supported.
        value: Reference to the column that should be used as a payload in the produced
            message in ``"plaintext"`` or ``"raw"`` format.
        headers: References to the table fields that must be provided as AMQP 1.0
            application properties (analogous to Kafka headers). Values are
            serialized as AMQP strings using their JSON representation, following
            the same encoding as :py:func:`pw.io.jsonlines.write` (e.g. ``42``
            for an int, ``"\"hello\""`` for a string, ``null`` for None,
            base64-encoded string for bytes). RabbitMQ Streams does not reliably
            confirm messages with non-string application property values, so all
            types are JSON-encoded. On the reader side, header values are
            available in ``_metadata.application_properties`` (when
            ``with_metadata=True``).
        name: A unique name for the connector. If provided, this name will be used in
            logs and monitoring dashboards.
        sort_by: If specified, the output will be sorted in ascending order based on the
            values of the given columns within each minibatch.
        tls_settings: TLS connection settings. Use ``TLSSettings`` to configure
            root certificates, client certificates, and verification mode.

    Examples: