Skip to content

RedisStreamParser

faststream.redis.parser.RedisStreamParser #

RedisStreamParser(*args: Any, **kwargs: Any)

Bases: _StreamParser

Source code in faststream/redis/parser/parsers.py 
150
151
152
153
154
155
156
157
def __init__(
    self,
    *args: Any,
    **kwargs: Any,
) -> None:
    super().__init__(*args, **kwargs)
    self.consumer: ConsumerProtocol = FAKE_CONSUMER
    self.group: str | None = None

pattern instance-attribute #

pattern = pattern

config instance-attribute #

config = config

consumer instance-attribute #

consumer: ConsumerProtocol = FAKE_CONSUMER

group instance-attribute #

group: str | None = None

msg_class class-attribute instance-attribute #

msg_class = RedisStreamMessage

parse_message async #

parse_message(
    message: Mapping[str, Any],
) -> StreamMessage[Mapping[str, Any]]
Source code in faststream/redis/parser/parsers.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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._build_message(
        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 
 97
 98
 99
100
101
async def decode_message(  # ruff: ignore[no-self-use]
    self,
    msg: "StreamMessage[Any]",
) -> DecodedMessage:
    return decode_message(msg)