Skip to content

Serialization

JSONSerializer

JSONSerializer

Default JSON serializer.

Supports: - dict/list → json.dumps/loads - Pydantic V2 models → model_validate_json/model_dump_json - dataclasses → asdict → json.dumps - str → encode to UTF-8 - bytes → pass through

By default the serializer raises on objects json cannot represent (e.g. datetime, Decimal) instead of silently coercing them via str(). Pass coerce_unknown_to_str=True to restore the legacy default=str coercion behaviour.

H14 — decoding into a stdlib dataclass does no type validation or coercion: a field declared qty: int silently receives whatever JSON type was actually present if the producer sent the wrong type. Unknown keys in the payload are dropped rather than raising. Use a Pydantic model as target_type (or a msgspec-based serializer) instead of a stdlib dataclass for untrusted input where wrong-typed fields must be rejected.

Source code in src/rabbitkit/serialization/json.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 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
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class JSONSerializer:
    """Default JSON serializer.

    Supports:
    - dict/list → json.dumps/loads
    - Pydantic V2 models → model_validate_json/model_dump_json
    - dataclasses → asdict → json.dumps
    - str → encode to UTF-8
    - bytes → pass through

    By default the serializer **raises** on objects ``json`` cannot represent
    (e.g. ``datetime``, ``Decimal``) instead of silently coercing them via
    ``str()``. Pass ``coerce_unknown_to_str=True`` to restore the legacy
    ``default=str`` coercion behaviour.

    H14 — decoding into a stdlib dataclass does **no type validation or
    coercion**: a field declared ``qty: int`` silently receives whatever
    JSON type was actually present if the producer sent the wrong type.
    Unknown keys in the payload are dropped rather than raising. Use a
    Pydantic model as ``target_type`` (or a msgspec-based serializer)
    instead of a stdlib dataclass for untrusted input where wrong-typed
    fields must be rejected.
    """

    #: M7: sane non-None default (64 MiB, matching
    #: ``CompressionMiddleware``'s ``max_decompressed_size`` default) so an
    #: uncompressed body is bounded out of the box. Pass ``max_parse_bytes=None``
    #: to opt out (unbounded) if you've already sized this elsewhere.
    _DEFAULT_MAX_PARSE_BYTES = 64 * 1024 * 1024

    def __init__(
        self,
        *,
        coerce_unknown_to_str: bool = False,
        max_parse_bytes: int | None = _DEFAULT_MAX_PARSE_BYTES,
    ) -> None:
        self._coerce = coerce_unknown_to_str
        # Defense-in-depth cap on the input size before json.loads (M7) — the
        # compression middleware already caps decompressed output; this
        # bounds the case where compression is off and a large body arrives
        # directly. Defaults to a sane non-None value rather than "off".
        self._max_parse_bytes = max_parse_bytes

    def _check_size(self, data: bytes) -> None:
        if self._max_parse_bytes is not None and len(data) > self._max_parse_bytes:
            raise ValueError(f"JSON input size {len(data)} exceeds max_parse_bytes={self._max_parse_bytes}")

    @property
    def content_type(self) -> str:
        return "application/json"

    def _default(self, obj: Any) -> Any:
        if self._coerce:
            return str(obj)
        raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")

    def encode(self, data: Any) -> bytes:
        """Serialize data to JSON bytes."""
        if isinstance(data, bytes):
            return data
        if isinstance(data, str):
            return data.encode("utf-8")

        # Pydantic V2 model
        if hasattr(data, "model_dump_json"):
            result = data.model_dump_json()
            return result.encode("utf-8") if isinstance(result, str) else result

        # Dataclass
        if is_dataclass(data) and not isinstance(data, type):
            return json.dumps(asdict(data), default=self._default).encode("utf-8")

        # Dict, list, or other JSON-serializable
        return json.dumps(data, default=self._default).encode("utf-8")

    def decode(self, data: bytes, target_type: type) -> Any:
        """Deserialize JSON bytes to target type."""
        self._check_size(data)
        if target_type is bytes:
            return data
        if target_type is str:
            return data.decode("utf-8")
        if target_type is dict:
            return json.loads(data)
        if target_type is list:
            return json.loads(data)

        # Pydantic V2 model
        if hasattr(target_type, "model_validate_json"):
            return target_type.model_validate_json(data)

        # Pydantic V2 model via model_validate (dict input)
        if hasattr(target_type, "model_validate"):
            parsed = json.loads(data)
            return target_type.model_validate(parsed)

        # Dataclass (H14: no type validation/coercion -- see DataclassDecoder
        # in serialization/pipeline.py for the full contract this mirrors.
        # Unknown keys are dropped rather than raising; a genuinely wrong
        # shape raises TypeError naming the target dataclass.)
        if is_dataclass(target_type):
            parsed = json.loads(data)
            if isinstance(parsed, dict):
                known_fields = {f.name for f in fields(target_type)}
                filtered = {k: v for k, v in parsed.items() if k in known_fields}
                try:
                    return target_type(**filtered)
                except TypeError as exc:
                    raise TypeError(f"Cannot decode into {target_type.__name__}: {exc}") from exc
            return parsed

        # Fallback: json.loads
        return json.loads(data)

Methods:

encode(data: Any) -> bytes

Serialize data to JSON bytes.

Source code in src/rabbitkit/serialization/json.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def encode(self, data: Any) -> bytes:
    """Serialize data to JSON bytes."""
    if isinstance(data, bytes):
        return data
    if isinstance(data, str):
        return data.encode("utf-8")

    # Pydantic V2 model
    if hasattr(data, "model_dump_json"):
        result = data.model_dump_json()
        return result.encode("utf-8") if isinstance(result, str) else result

    # Dataclass
    if is_dataclass(data) and not isinstance(data, type):
        return json.dumps(asdict(data), default=self._default).encode("utf-8")

    # Dict, list, or other JSON-serializable
    return json.dumps(data, default=self._default).encode("utf-8")

decode(data: bytes, target_type: type) -> Any

Deserialize JSON bytes to target type.

Source code in src/rabbitkit/serialization/json.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def decode(self, data: bytes, target_type: type) -> Any:
    """Deserialize JSON bytes to target type."""
    self._check_size(data)
    if target_type is bytes:
        return data
    if target_type is str:
        return data.decode("utf-8")
    if target_type is dict:
        return json.loads(data)
    if target_type is list:
        return json.loads(data)

    # Pydantic V2 model
    if hasattr(target_type, "model_validate_json"):
        return target_type.model_validate_json(data)

    # Pydantic V2 model via model_validate (dict input)
    if hasattr(target_type, "model_validate"):
        parsed = json.loads(data)
        return target_type.model_validate(parsed)

    # Dataclass (H14: no type validation/coercion -- see DataclassDecoder
    # in serialization/pipeline.py for the full contract this mirrors.
    # Unknown keys are dropped rather than raising; a genuinely wrong
    # shape raises TypeError naming the target dataclass.)
    if is_dataclass(target_type):
        parsed = json.loads(data)
        if isinstance(parsed, dict):
            known_fields = {f.name for f in fields(target_type)}
            filtered = {k: v for k, v in parsed.items() if k in known_fields}
            try:
                return target_type(**filtered)
            except TypeError as exc:
                raise TypeError(f"Cannot decode into {target_type.__name__}: {exc}") from exc
        return parsed

    # Fallback: json.loads
    return json.loads(data)

MsgspecSerializer

MsgspecSerializer

High-performance serializer using msgspec.

Requires msgspec to be installed (optional dependency). Falls back with clear error if not available.

M7: caps the input size before decoding (64 MiB by default, matching CompressionMiddleware's max_decompressed_size default) — without it, a large uncompressed body is fully materialized with no bound. Pass max_parse_bytes=None to opt out.

Source code in src/rabbitkit/serialization/msgspec.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 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
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class MsgspecSerializer:
    """High-performance serializer using msgspec.

    Requires msgspec to be installed (optional dependency).
    Falls back with clear error if not available.

    M7: caps the input size before decoding (64 MiB by default, matching
    ``CompressionMiddleware``'s ``max_decompressed_size`` default) — without
    it, a large uncompressed body is fully materialized with no bound.
    Pass ``max_parse_bytes=None`` to opt out.
    """

    #: M7: see JSONSerializer's identical default for the rationale.
    _DEFAULT_MAX_PARSE_BYTES = 64 * 1024 * 1024

    def __init__(self, *, max_parse_bytes: int | None = _DEFAULT_MAX_PARSE_BYTES) -> None:
        try:
            import msgspec

            self._msgspec = msgspec
            self._encoder = msgspec.json.Encoder()
            self._decoder = msgspec.json.Decoder()
            self._decoders: dict[Any, Any] = {}  # cached decoders per target_type (perf)
        except ImportError as e:  # pragma: no cover
            raise ImportError(  # pragma: no cover
                "msgspec is required for MsgspecSerializer. Install it with: pip install rabbitkit[msgspec]"
            ) from e
        self._max_parse_bytes = max_parse_bytes

    def _check_size(self, data: bytes) -> None:
        if self._max_parse_bytes is not None and len(data) > self._max_parse_bytes:
            raise ValueError(f"JSON input size {len(data)} exceeds max_parse_bytes={self._max_parse_bytes}")

    @property
    def content_type(self) -> str:
        """Advisory only (M10): this is the ``content_type`` used when
        *publishing* (set on the outgoing AMQP message property). It is
        **not** verified against an incoming message's declared
        ``content_type`` on :meth:`decode` — decode is driven solely by the
        handler's declared parameter type, matching every other built-in
        serializer in this codebase (none of them negotiate/verify
        content-type on the consume side). If a message's actual body
        doesn't match what this serializer expects (e.g. it's not JSON at
        all), :meth:`decode` raises with a message naming both the target
        type and a content-type-mismatch hint, rather than a raw
        ``msgspec.DecodeError``.
        """
        return "application/json"

    def encode(self, data: Any) -> bytes:
        """Serialize data to JSON bytes using msgspec."""
        if isinstance(data, bytes):
            return data
        if isinstance(data, str):
            return data.encode("utf-8")

        # msgspec Struct
        if isinstance(data, self._msgspec.Struct):
            return self._msgspec.json.encode(data)

        # General encoding
        return self._encoder.encode(data)

    def decode(self, data: bytes, target_type: type) -> Any:
        """Deserialize JSON bytes using msgspec.

        M10: raises with a clearer message (naming the target type and
        hinting at a content-type mismatch) instead of a raw
        ``msgspec.DecodeError`` if the body isn't valid JSON for
        *target_type* — most commonly caused by a message whose actual
        content_type doesn't match what this serializer expects (see
        :attr:`content_type`'s docstring — this is never verified upfront).
        """
        try:
            return self._decode(data, target_type)
        except self._msgspec.DecodeError as exc:
            raise ValueError(
                f"Failed to decode message body as JSON for target type {target_type!r}: {exc}. "
                "If the message's actual content_type doesn't match what this serializer expects "
                "(application/json), that's the likely cause — MsgspecSerializer does not verify "
                "content_type before decoding."
            ) from exc

    def _decode(self, data: bytes, target_type: type) -> Any:
        self._check_size(data)
        if target_type is bytes:
            return data
        if target_type is str:
            return data.decode("utf-8")
        if target_type is dict or target_type is list:
            return self._msgspec.json.decode(data)

        # Generic alias (dict[str, Any], list[int], …) — origin is dict/list
        origin = getattr(target_type, "__origin__", None)
        if origin is dict or origin is list:
            decoder = self._decoders.get(target_type)
            if decoder is None:
                try:
                    decoder = self._msgspec.json.Decoder(type=target_type)
                except Exception:
                    return self._msgspec.json.decode(data)
                self._decoders[target_type] = decoder
            return decoder.decode(data)

        # Pydantic V2 model — model_validate_json is faster than json.loads + model_validate
        if hasattr(target_type, "model_validate_json"):
            return target_type.model_validate_json(data)

        # msgspec Struct — fastest path for msgspec-native types
        try:
            is_struct = issubclass(target_type, self._msgspec.Struct)
        except TypeError:
            is_struct = False
        if is_struct:
            return self._msgspec.json.decode(data, type=target_type)

        # General typed decoder — cached per target_type (Decoder(type=T) codegens
        # a converter; rebuilding per call defeats msgspec's performance advantage).
        decoder = self._decoders.get(target_type)
        if decoder is None:
            try:
                decoder = self._msgspec.json.Decoder(type=target_type)
            except Exception:
                return self._msgspec.json.decode(data)
            self._decoders[target_type] = decoder
        return decoder.decode(data)

Attributes

content_type: str property

Advisory only (M10): this is the content_type used when publishing (set on the outgoing AMQP message property). It is not verified against an incoming message's declared content_type on :meth:decode — decode is driven solely by the handler's declared parameter type, matching every other built-in serializer in this codebase (none of them negotiate/verify content-type on the consume side). If a message's actual body doesn't match what this serializer expects (e.g. it's not JSON at all), :meth:decode raises with a message naming both the target type and a content-type-mismatch hint, rather than a raw msgspec.DecodeError.

Methods:

encode(data: Any) -> bytes

Serialize data to JSON bytes using msgspec.

Source code in src/rabbitkit/serialization/msgspec.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
def encode(self, data: Any) -> bytes:
    """Serialize data to JSON bytes using msgspec."""
    if isinstance(data, bytes):
        return data
    if isinstance(data, str):
        return data.encode("utf-8")

    # msgspec Struct
    if isinstance(data, self._msgspec.Struct):
        return self._msgspec.json.encode(data)

    # General encoding
    return self._encoder.encode(data)

decode(data: bytes, target_type: type) -> Any

Deserialize JSON bytes using msgspec.

M10: raises with a clearer message (naming the target type and hinting at a content-type mismatch) instead of a raw msgspec.DecodeError if the body isn't valid JSON for target_type — most commonly caused by a message whose actual content_type doesn't match what this serializer expects (see :attr:content_type's docstring — this is never verified upfront).

Source code in src/rabbitkit/serialization/msgspec.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def decode(self, data: bytes, target_type: type) -> Any:
    """Deserialize JSON bytes using msgspec.

    M10: raises with a clearer message (naming the target type and
    hinting at a content-type mismatch) instead of a raw
    ``msgspec.DecodeError`` if the body isn't valid JSON for
    *target_type* — most commonly caused by a message whose actual
    content_type doesn't match what this serializer expects (see
    :attr:`content_type`'s docstring — this is never verified upfront).
    """
    try:
        return self._decode(data, target_type)
    except self._msgspec.DecodeError as exc:
        raise ValueError(
            f"Failed to decode message body as JSON for target type {target_type!r}: {exc}. "
            "If the message's actual content_type doesn't match what this serializer expects "
            "(application/json), that's the likely cause — MsgspecSerializer does not verify "
            "content_type before decoding."
        ) from exc

SerializationPipeline

SerializationPipeline

Two-stage serialization. Implements Serializer protocol.

Source code in src/rabbitkit/serialization/pipeline.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class SerializationPipeline:
    """Two-stage serialization. Implements Serializer protocol."""

    def __init__(self, parser: MessageParser, decoder: MessageDecoder) -> None:
        self._parser = parser
        self._decoder = decoder

    def encode(self, data: Any) -> bytes:
        intermediate = self._decoder.encode(data)
        return self._parser.serialize(intermediate)

    def decode(self, data: bytes, target_type: type) -> Any:
        intermediate = self._parser.parse(data)
        return self._decoder.decode(intermediate, target_type)

    @property
    def content_type(self) -> str:
        return self._parser.content_type

Parsers & Decoders

JsonParser

Built-in JSON parser.

By default serialize raises on objects json cannot represent (e.g. datetime, Decimal) rather than silently coercing them via str(). Pass coerce_unknown_to_str=True to restore the legacy default=str coercion behaviour.

M7: caps the input size before json.loads (64 MiB by default, matching CompressionMiddleware's max_decompressed_size default) -- without it, a large uncompressed body is fully materialized with no bound. Pass max_parse_bytes=None to opt out.

Source code in src/rabbitkit/serialization/pipeline.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class JsonParser:
    """Built-in JSON parser.

    By default ``serialize`` **raises** on objects ``json`` cannot represent
    (e.g. ``datetime``, ``Decimal``) rather than silently coercing them via
    ``str()``. Pass ``coerce_unknown_to_str=True`` to restore the legacy
    ``default=str`` coercion behaviour.

    M7: caps the input size before ``json.loads`` (64 MiB by default,
    matching ``CompressionMiddleware``'s ``max_decompressed_size`` default)
    -- without it, a large uncompressed body is fully materialized with no
    bound. Pass ``max_parse_bytes=None`` to opt out.
    """

    #: M7: see JSONSerializer's identical default for the rationale.
    _DEFAULT_MAX_PARSE_BYTES = 64 * 1024 * 1024

    def __init__(
        self,
        *,
        coerce_unknown_to_str: bool = False,
        max_parse_bytes: int | None = _DEFAULT_MAX_PARSE_BYTES,
    ) -> None:
        self._coerce = coerce_unknown_to_str
        self._max_parse_bytes = max_parse_bytes

    def _check_size(self, data: bytes) -> None:
        if self._max_parse_bytes is not None and len(data) > self._max_parse_bytes:
            raise ValueError(f"JSON input size {len(data)} exceeds max_parse_bytes={self._max_parse_bytes}")

    def parse(self, data: bytes, content_type: str | None = None) -> Any:
        self._check_size(data)
        return json.loads(data)

    def _default(self, data: Any) -> Any:
        if self._coerce:
            return str(data)
        raise TypeError(f"Object of type {type(data).__name__} is not JSON serializable")

    def serialize(self, data: Any) -> bytes:
        return json.dumps(data, default=self._default).encode("utf-8")

    @property
    def content_type(self) -> str:
        return "application/json"

PydanticDecoder

Decoder that uses Pydantic model_validate for decoding.

Companion to DataclassDecoder's H14 note: this is the decoder that note recommends for untrusted input, on the premise that model_validate rejects wrong-shaped data instead of silently accepting it. That premise used to be undermined by an isinstance(data, dict) guard here -- a non-dict top-level payload (a producer sending a JSON array/string/number instead of an object, whether by bug or by an attacker probing for a decoder that skips validation on the untaken branch) bypassed model_validate entirely and returned the raw, un-validated value straight to the handler. Pydantic already raises a clean ValidationError for a non-dict input (model_type error), so there is no reason to special-case it -- doing so only suppressed the exact validation this decoder exists to provide.

Source code in src/rabbitkit/serialization/pipeline.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class PydanticDecoder:
    """Decoder that uses Pydantic model_validate for decoding.

    Companion to ``DataclassDecoder``'s H14 note: this is the decoder that
    note recommends for untrusted input, on the premise that
    ``model_validate`` rejects wrong-shaped data instead of silently
    accepting it. That premise used to be undermined by an
    ``isinstance(data, dict)`` guard here -- a non-dict top-level payload
    (a producer sending a JSON array/string/number instead of an object,
    whether by bug or by an attacker probing for a decoder that skips
    validation on the untaken branch) bypassed ``model_validate`` entirely
    and returned the raw, un-validated value straight to the handler.
    Pydantic already raises a clean ``ValidationError`` for a non-dict
    input (``model_type`` error), so there is no reason to special-case
    it -- doing so only suppressed the exact validation this decoder
    exists to provide.
    """

    def decode(self, data: Any, target_type: type) -> Any:
        if hasattr(target_type, "model_validate"):
            return target_type.model_validate(data)
        return data

    def encode(self, data: Any) -> Any:
        if hasattr(data, "model_dump"):
            return data.model_dump()
        return data

DataclassDecoder

Decoder for stdlib dataclasses.

H14 — no type validation or coercion: unlike PydanticDecoder, this does NOT check field types. A field declared qty: int silently receives whatever JSON type was actually present (e.g. the string "3") if the producer sent the wrong type — stdlib dataclasses perform no runtime type checking on construction. Use PydanticDecoder (or a msgspec-based decoder) instead for untrusted input where wrong-typed fields must be rejected.

Unknown keys in the incoming dict (not a declared field) are silently dropped rather than raising TypeError — this keeps a producer that adds a new field (forward-compatible) from turning every message into a decode failure (misclassified PERMANENT, straight to the DLQ) until every consumer is upgraded. A genuinely wrong shape (a missing required field, etc.) still raises — as a TypeError naming the target dataclass, not a bare constructor traceback.

Source code in src/rabbitkit/serialization/pipeline.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
class DataclassDecoder:
    """Decoder for stdlib dataclasses.

    H14 — no type validation or coercion: unlike ``PydanticDecoder``, this
    does NOT check field types. A field declared ``qty: int`` silently
    receives whatever JSON type was actually present (e.g. the string
    ``"3"``) if the producer sent the wrong type — stdlib dataclasses
    perform no runtime type checking on construction. **Use
    ``PydanticDecoder`` (or a msgspec-based decoder) instead for untrusted
    input where wrong-typed fields must be rejected.**

    Unknown keys in the incoming dict (not a declared field) are silently
    dropped rather than raising ``TypeError`` — this keeps a producer that
    adds a new field (forward-compatible) from turning every message into a
    decode failure (misclassified PERMANENT, straight to the DLQ) until
    every consumer is upgraded. A genuinely wrong shape (a missing required
    field, etc.) still raises — as a ``TypeError`` naming the target
    dataclass, not a bare constructor traceback.
    """

    def decode(self, data: Any, target_type: type) -> Any:
        import dataclasses

        if dataclasses.is_dataclass(target_type) and isinstance(data, dict):
            known_fields = {f.name for f in dataclasses.fields(target_type)}
            filtered = {k: v for k, v in data.items() if k in known_fields}
            try:
                return target_type(**filtered)
            except TypeError as exc:
                raise TypeError(f"Cannot decode into {target_type.__name__}: {exc}") from exc
        return data

    def encode(self, data: Any) -> Any:
        import dataclasses

        if dataclasses.is_dataclass(data) and not isinstance(data, type):
            return dataclasses.asdict(data)
        return data

RawDecoder

Pass-through decoder — no transformation.

Source code in src/rabbitkit/serialization/pipeline.py 
248
249
250
251
252
253
254
255
class RawDecoder:
    """Pass-through decoder — no transformation."""

    def decode(self, data: Any, target_type: type) -> Any:
        return data

    def encode(self, data: Any) -> Any:
        return data

MessageParser

Bases: Protocol

Stage 1: raw bytes → intermediate form.

Source code in src/rabbitkit/serialization/pipeline.py 
 96
 97
 98
 99
100
101
102
class MessageParser(Protocol):
    """Stage 1: raw bytes → intermediate form."""

    def parse(self, data: bytes, content_type: str | None = None) -> Any: ...
    def serialize(self, data: Any) -> bytes: ...
    @property
    def content_type(self) -> str: ...

MessageDecoder

Bases: Protocol

Stage 2: intermediate → typed Python object.

Source code in src/rabbitkit/serialization/pipeline.py 
105
106
107
108
109
class MessageDecoder(Protocol):
    """Stage 2: intermediate → typed Python object."""

    def decode(self, data: Any, target_type: type) -> Any: ...
    def encode(self, data: Any) -> Any: ...