Skip to content

JSONMessageFormat

faststream.redis.parser.json.JSONMessageFormat #

JSONMessageFormat(data, headers=None)

Bases: MessageFormat

Message format to encode into JSON and parse it.

Source code in faststream/redis/parser/message.py 
21
22
23
24
25
26
27
def __init__(
    self,
    data: bytes,
    headers: dict[str, Any] | None = None,
) -> None:
    self.data = data
    self.headers = headers or {}

data instance-attribute #

data = data

headers instance-attribute #

headers = headers or {}

encode classmethod #

encode(
    *,
    message,
    reply_to,
    headers,
    correlation_id,
    serializer=None,
)
Source code in faststream/redis/parser/json.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@classmethod
def encode(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: str | None,
    headers: dict[str, Any] | None,
    correlation_id: str,
    serializer: Optional["SerializerProto"] = None,
) -> bytes:
    msg = cls.build(
        message=message,
        reply_to=reply_to,
        headers=headers,
        correlation_id=correlation_id,
        serializer=serializer,
    )
    return dump_json({
        "data": msg.data,
        "headers": msg.headers,
    })

parse classmethod #

parse(data)
Source code in faststream/redis/parser/json.py 
39
40
41
42
43
44
45
46
47
48
49
50
@classmethod
def parse(cls, data: bytes) -> tuple[bytes, dict[str, Any]]:
    headers: dict[str, Any]
    try:
        parsed_data = json_loads(data)
        final_data = parsed_data["data"].encode()
        headers = parsed_data.get("headers", {})
    except Exception:
        # Raw Redis message format
        final_data = data
        headers = {}
    return final_data, headers

build classmethod #

build(
    *,
    message,
    reply_to,
    headers,
    correlation_id,
    serializer=None,
)
Source code in faststream/redis/parser/message.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@classmethod
def build(
    cls,
    *,
    message: Union[Sequence["SendableMessage"], "SendableMessage"],
    reply_to: str | None,
    headers: dict[str, Any] | None,
    correlation_id: str,
    serializer: Optional["SerializerProto"] = None,
) -> "MessageFormat":
    payload, content_type = encode_message(message, serializer=serializer)

    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,
    )