Skip to content

RedisBatchStreamParser

faststream.redis.parser.RedisBatchStreamParser #

RedisBatchStreamParser(
    config: ParserConfig = EMPTY,
    pattern: Optional[Pattern[str]] = None,
)

Bases: SimpleParser

Source code in faststream/redis/parser/parsers.py 
39
40
41
42
43
44
45
def __init__(
    self,
    config: "ParserConfig" = EMPTY,
    pattern: Optional["Pattern[str]"] = None,
) -> None:
    self.pattern = pattern
    self.config = config

pattern instance-attribute #

pattern = pattern

config instance-attribute #

config = config

msg_class class-attribute instance-attribute #

msg_class = RedisBatchStreamMessage

parse_message async #

parse_message(
    message: Mapping[str, Any],
) -> StreamMessage[Mapping[str, Any]]
Source code in faststream/redis/parser/parsers.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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: Mapping[str, Any]) -> dict[str, Any]
Source code in faststream/redis/parser/parsers.py 
73
74
75
76
77
78
79
80
81
def get_path(self, message: Mapping[str, Any]) -> dict[str, Any]:
    if (
        (path_re := self.pattern)
        and message.get("pattern")
        and (match := path_re.match(message["channel"]))
    ):
        return match.groupdict()

    return {}

decode_message async #

decode_message(msg: StreamMessage[Any]) -> DecodedMessage
Source code in faststream/redis/parser/parsers.py 
83
84
85
86
87
async def decode_message(
    self,
    msg: "StreamMessage[Any]",
) -> DecodedMessage:
    return decode_message(msg)