Skip to content

RedisBatchStreamParser

faststream.redis.parser.RedisBatchStreamParser #

RedisBatchStreamParser(pattern=None)

Bases: SimpleParser

Source code in faststream/redis/parser.py 
129
130
131
132
133
def __init__(
    self,
    pattern: Optional["Pattern[str]"] = None,
) -> None:
    self.pattern = pattern

pattern instance-attribute #

pattern = pattern

msg_class class-attribute instance-attribute #

msg_class = RedisBatchStreamMessage

parse_message async #

parse_message(message)
Source code in faststream/redis/parser.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def parse_message(
    self,
    message: Mapping[str, Any],
) -> "StreamMessage[Mapping[str, Any]]":
    data, headers, batch_headers = self._parse_data(message)

    id_ = gen_cor_id()

    return self.msg_class(
        raw_message=message,
        body=data,
        path=self.get_path(message),
        headers=headers,
        batch_headers=batch_headers,
        reply_to=headers.get("reply_to", ""),
        content_type=headers.get("content-type"),
        message_id=headers.get("message_id", id_),
        correlation_id=headers.get("correlation_id", id_),
    )

get_path #

get_path(message)
Source code in faststream/redis/parser.py 
161
162
163
164
165
166
167
168
169
170
def get_path(self, message: Mapping[str, Any]) -> "AnyDict":
    if (
        message.get("pattern")
        and (path_re := self.pattern)
        and (match := path_re.match(message["channel"]))
    ):
        return match.groupdict()

    else:
        return {}

decode_message async #

decode_message(msg)
Source code in faststream/redis/parser.py 
172
173
174
175
176
async def decode_message(
    self,
    msg: "StreamMessage[MsgType]",
) -> DecodedMessage:
    return decode_message(msg)