Skip to content

KafkaResponse

faststream.confluent.response.KafkaResponse #

KafkaResponse(
    body: SendableMessage,
    *,
    headers: dict[str, Any] | None = None,
    correlation_id: str | None = None,
    timestamp_ms: int | None = None,
    key: bytes | str | None = None,
)

Bases: Response

Source code in faststream/confluent/response.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def __init__(
    self,
    body: "SendableMessage",
    *,
    headers: dict[str, Any] | None = None,
    correlation_id: str | None = None,
    timestamp_ms: int | None = None,
    key: bytes | str | None = None,
) -> None:
    super().__init__(
        body=body,
        headers=headers,
        correlation_id=correlation_id,
    )

    self.timestamp_ms = timestamp_ms
    self.key = key

timestamp_ms instance-attribute #

timestamp_ms = timestamp_ms

key instance-attribute #

key = key

body instance-attribute #

body = body

headers instance-attribute #

headers = headers or {}

correlation_id instance-attribute #

correlation_id = correlation_id

as_publish_command #

as_publish_command() -> KafkaPublishCommand
Source code in faststream/confluent/response.py 
31
32
33
34
35
36
37
38
39
40
41
42
@override
def as_publish_command(self) -> "KafkaPublishCommand":
    return KafkaPublishCommand(
        self.body,
        headers=self.headers,
        correlation_id=self.correlation_id,
        _publish_type=PublishType.PUBLISH,
        # Kafka specific
        topic="",
        key=self.key,
        timestamp_ms=self.timestamp_ms,
    )