Skip to content

RawMessage

faststream.redis.parser.RawMessage #

Bases: BaseModel

data instance-attribute #

data: bytes

headers class-attribute instance-attribute #

headers: AnyDict = Field(default_factory=dict)

build classmethod #

build(
    message: SendableMessage,
    reply_to: str = "",
    headers: Optional[AnyDict] = None,
    correlation_id: Optional[str] = None,
) -> RawMessage
Source code in faststream/redis/parser.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@classmethod
def build(
    cls,
    message: SendableMessage,
    reply_to: str = "",
    headers: Optional[AnyDict] = None,
    correlation_id: Optional[str] = None,
) -> "RawMessage":
    payload, content_type = encode_message(message)

    headers_to_send = {
        "correlation_id": correlation_id or str(uuid4()),
    }

    if content_type:
        headers_to_send["content-type"] = content_type

    if reply_to:
        headers_to_send["reply_to"] = reply_to

    if headers is not None:
        headers_to_send.update(headers)

    return cls(
        data=payload,
        headers=headers_to_send,
    )

encode classmethod #

encode(
    message: SendableMessage,
    reply_to: str = "",
    headers: Optional[AnyDict] = None,
    correlation_id: Optional[str] = None,
) -> str
Source code in faststream/redis/parser.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@classmethod
def encode(
    cls,
    message: SendableMessage,
    reply_to: str = "",
    headers: Optional[AnyDict] = None,
    correlation_id: Optional[str] = None,
) -> str:
    return model_to_json(
        cls.build(
            message=message,
            reply_to=reply_to,
            headers=headers,
            correlation_id=correlation_id,
        )
    )