Skip to content

SimpleParser

faststream.redis.parser.parsers.SimpleParser #

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

msg_class instance-attribute #

msg_class: type[StreamMessage[Any]]

pattern instance-attribute #

pattern = pattern

config instance-attribute #

config = config

parse_message async #

parse_message(
    message: Mapping[str, Any],
) -> StreamMessage[Mapping[str, Any]]
Source code in faststream/redis/parser/parsers.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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,
        # Only pattern-subscribed messages have "pattern" set;
        # guard here before calling match_path.
        path=match_path(self.pattern, message["channel"])
        if message.get("pattern")
        else {},
        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_),
    )

decode_message async #

decode_message(msg: StreamMessage[Any]) -> DecodedMessage
Source code in faststream/redis/parser/parsers.py 
78
79
80
81
82
async def decode_message(
    self,
    msg: "StreamMessage[Any]",
) -> DecodedMessage:
    return decode_message(msg)