Skip to content

MessageFormat

faststream.redis.parser.MessageFormat #

MessageFormat(data, headers=None)

Bases: ABC

A class to represent a raw Redis message.

Source code in faststream/redis/parser.py 
115
116
117
118
119
120
121
def __init__(
    self,
    data: bytes,
    headers: Optional["AnyDict"] = None,
) -> None:
    self.data = data
    self.headers = headers or {}

data instance-attribute #

data = data

headers instance-attribute #

headers = headers or {}

build classmethod #

build(*, message, reply_to, headers, correlation_id)
Source code in faststream/redis/parser.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@classmethod
def build(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: Optional[str],
    headers: Optional["AnyDict"],
    correlation_id: str,
) -> "MessageFormat":
    payload, content_type = encode_message(message)

    headers_to_send = {
        "correlation_id": correlation_id,
    }

    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 abstractmethod classmethod #

encode(*, message, reply_to, headers, correlation_id)
Source code in faststream/redis/parser.py 
152
153
154
155
156
157
158
159
160
161
162
@classmethod
@abstractmethod
def encode(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: Optional[str],
    headers: Optional["AnyDict"],
    correlation_id: str,
) -> bytes:
    raise NotImplementedError()

parse abstractmethod classmethod #

parse(data)
Source code in faststream/redis/parser.py 
164
165
166
167
@classmethod
@abstractmethod
def parse(cls, data: bytes) -> Tuple[bytes, "AnyDict"]:
    raise NotImplementedError()