Skip to content

encode_message

faststream.message.encode_message #

encode_message(
    msg: Union[Sequence[SendableMessage], SendableMessage],
    serializer: Optional[SerializerProto],
) -> tuple[bytes, str | None]

Encodes a message.

Source code in faststream/message/utils.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def encode_message(
    msg: Union[Sequence["SendableMessage"], "SendableMessage"],
    serializer: Optional["SerializerProto"],
) -> tuple[bytes, str | None]:
    """Encodes a message."""
    if msg is None:
        return (
            b"",
            None,
        )

    if isinstance(msg, bytes):
        return (
            msg,
            None,
        )

    if isinstance(msg, str):
        return (
            msg.encode(),
            ContentTypes.TEXT.value,
        )

    if serializer is not None:
        return (
            serializer.encode(msg),
            ContentTypes.JSON.value,
        )

    return (
        json_dumps(msg),
        ContentTypes.JSON.value,
    )