acemq_amqp.codec

  1# Copyright 2026 AceMQ.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     https://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""Turning a payload into bytes, and bytes back into a payload.
 16
 17The content type is what makes this work across languages: a Java producer says
 18``application/json`` and a Python consumer picks the codec that answers for it,
 19without either end having been told about the other. A message that arrives with
 20no content type at all is the interesting case, and the rule is the same one the
 21.NET library uses — nothing has been ruled out, so every codec is a candidate
 22and the first that can actually read the body wins.
 23
 24Decoding returns the value rather than filling a destination that was passed in,
 25which is where this departs from Go. Go's ``Decode`` takes a pointer because
 26that is how every Go decoder works, and it then needs a second interface to get
 27the content type to a composite codec that has to choose. Python can put the
 28content type on ``decode`` as a defaulted argument, so one method serves both the
 29codecs that need it and the codecs that do not.
 30"""
 31
 32from __future__ import annotations
 33
 34import dataclasses
 35import json
 36import threading
 37from collections.abc import Callable
 38from typing import Any, Protocol, runtime_checkable
 39
 40from .ack import FatalError
 41
 42#: What :class:`JsonCodec` writes, and what Java, Go and .NET write.
 43JSON_CONTENT_TYPE = "application/json"
 44
 45#: What :class:`BytesCodec` writes when nothing better is known.
 46BYTES_CONTENT_TYPE = "application/octet-stream"
 47
 48#: What :class:`TextCodec` writes.
 49TEXT_CONTENT_TYPE = "text/plain; charset=utf-8"
 50
 51
 52@runtime_checkable
 53class Codec(Protocol):
 54    """Reads and writes message bodies in one format."""
 55
 56    @property
 57    def content_type(self) -> str:
 58        """What this codec writes onto a message it encodes."""
 59
 60    def encode(self, payload: Any) -> bytes:
 61        """Turns a payload into bytes.
 62
 63        :param payload: what the application wants to send
 64        :returns: the body
 65        :raises TypeError: when the payload is not something this codec writes
 66        """
 67
 68    def decode(self, body: bytes, content_type: str | None = None) -> Any:
 69        """Turns bytes back into a payload.
 70
 71        :param body: the message body
 72        :param content_type: what the sender said it was, or ``None`` when the
 73            sender said nothing
 74        :returns: the payload
 75        :raises FatalError: when the body cannot be read, which is fatal because
 76            the same bytes will not read any better on a second attempt
 77        """
 78
 79    def can_decode(self, content_type: str | None) -> bool:
 80        """Whether this codec should be offered a message.
 81
 82        :param content_type: the sender's content type, or ``None`` when there
 83            was none
 84        :returns: whether this codec claims it
 85        """
 86
 87
 88class JsonCodec:
 89    """Reads and writes JSON. The default, and the format every AceMQ library
 90    has without an extra dependency.
 91
 92    Dataclasses are encoded through :func:`dataclasses.asdict`, so a payload
 93    written as a dataclass goes on the wire as an object with its field names.
 94    Everything else is handed to :mod:`json` unchanged, which covers dicts,
 95    lists and the scalars.
 96
 97    Decoding produces dicts and lists rather than a class, because the wire says
 98    nothing about which class was meant. An application that wants its own type
 99    back constructs it from what it gets, where it can decide what a missing
100    field means.
101    """
102
103    @property
104    def content_type(self) -> str:
105        return JSON_CONTENT_TYPE
106
107    def encode(self, payload: Any) -> bytes:
108        if dataclasses.is_dataclass(payload) and not isinstance(payload, type):
109            payload = dataclasses.asdict(payload)
110        try:
111            return json.dumps(payload).encode("utf-8")
112        except TypeError as failure:
113            raise TypeError(
114                f"acemq: a {type(payload).__name__} is not something JSON can carry: {failure}"
115            ) from failure
116
117    def decode(self, body: bytes, content_type: str | None = None) -> Any:
118        try:
119            return json.loads(body)
120        except (ValueError, UnicodeDecodeError) as failure:
121            # Fatal rather than retryable: a body that is not JSON is not JSON
122            # the next time either, and retrying it only holds a queue open
123            # until the message ages out.
124            raise FatalError(f"acemq: this message is not JSON: {failure}") from failure
125
126    def can_decode(self, content_type: str | None) -> bool:
127        """Accepts JSON, any ``+json`` media type, and a message with none.
128
129        Answering for a message whose sender set no content type is deliberate
130        and is what the other libraries do: JSON is the default format, so an
131        untyped message is far more likely to be JSON than anything else, and
132        something has to read it.
133        """
134        if content_type is None or content_type == "":
135            return True
136        lowered = content_type.lower()
137        return (
138            lowered.startswith(JSON_CONTENT_TYPE)
139            or lowered.startswith("text/json")
140            or "+json" in lowered
141        )
142
143
144class BytesCodec:
145    """Passes bodies through without touching them.
146
147    For a payload that is already encoded — an image, something another system's
148    serializer produced — and for reading a message this process has no type
149    for. Replaying a dead-lettered message uses it: the bytes that were
150    committed are the bytes that should go back, and re-encoding through a class
151    that has since gained a field would produce something else.
152
153    It answers for every content type, which is why it has to be asked for
154    rather than found: put it last in a :class:`CompositeCodec`, or nothing
155    after it is ever reached.
156    """
157
158    @property
159    def content_type(self) -> str:
160        return BYTES_CONTENT_TYPE
161
162    def encode(self, payload: Any) -> bytes:
163        if payload is None:
164            return b""
165        if isinstance(payload, bytes):
166            return payload
167        if isinstance(payload, bytearray | memoryview):
168            return bytes(payload)
169        if isinstance(payload, str):
170            return payload.encode("utf-8")
171        raise TypeError(
172            f"acemq: BytesCodec cannot encode a {type(payload).__name__}; "
173            "it takes bytes or a string"
174        )
175
176    def decode(self, body: bytes, content_type: str | None = None) -> Any:
177        return bytes(body)
178
179    def can_decode(self, content_type: str | None) -> bool:
180        return True
181
182
183class TextCodec:
184    """Reads and writes text.
185
186    For messages that really are text — a line of a log, a command somebody
187    typed — rather than a structure that happens to be readable. Prefer
188    :class:`JsonCodec` for anything with fields.
189    """
190
191    @property
192    def content_type(self) -> str:
193        return TEXT_CONTENT_TYPE
194
195    def encode(self, payload: Any) -> bytes:
196        if isinstance(payload, str):
197            return payload.encode("utf-8")
198        if isinstance(payload, bytes):
199            return payload
200        raise TypeError(
201            f"acemq: TextCodec cannot encode a {type(payload).__name__}; "
202            "it takes a string or bytes"
203        )
204
205    def decode(self, body: bytes, content_type: str | None = None) -> Any:
206        try:
207            return body.decode("utf-8")
208        except UnicodeDecodeError as failure:
209            raise FatalError(f"acemq: this message is not UTF-8 text: {failure}") from failure
210
211    def can_decode(self, content_type: str | None) -> bool:
212        """Accepts ``text/*`` and nothing else.
213
214        Unlike :class:`JsonCodec` it does not answer for a message with no
215        content type, because there is no reason to think an untyped message is
216        text — and a codec that decodes anything into a string never fails, so
217        it would quietly hand a handler the printable rendering of a body it
218        should have refused.
219        """
220        return content_type is not None and content_type.lower().startswith("text/")
221
222
223class CompositeCodec:
224    """Picks a codec by the message's content type.
225
226    For a queue carrying more than one format — during a migration, or where
227    several producers were written at different times::
228
229        codec = CompositeCodec(JsonCodec(), TextCodec())
230
231    The first codec is what it writes. All of them are offered a message to
232    read, in the order they were given, and the first that claims the content
233    type *and* succeeds wins. Order matters when two overlap: :class:`BytesCodec`
234    answers for everything, so anything after it would never be reached.
235    """
236
237    def __init__(self, *codecs: Codec) -> None:
238        if not codecs:
239            raise ValueError("acemq: a CompositeCodec needs at least one codec in it")
240        self._codecs: tuple[Codec, ...] = codecs
241
242    @property
243    def codecs(self) -> tuple[Codec, ...]:
244        """The codecs, in the order they are tried."""
245        return self._codecs
246
247    @property
248    def content_type(self) -> str:
249        """The first codec's, since that is the one that encodes."""
250        return self._codecs[0].content_type
251
252    def encode(self, payload: Any) -> bytes:
253        return self._codecs[0].encode(payload)
254
255    def decode(self, body: bytes, content_type: str | None = None) -> Any:
256        """Reads a body with whichever codec recognises the content type.
257
258        With no content type nothing can be ruled out, so every codec is a
259        candidate rather than the first being assumed right. Candidates are
260        tried in turn and the first that reads the body wins — so a message that
261        claims one format and is written in another is still read — and only
262        when every one has refused is anything raised, carrying all the reasons
263        rather than just the last.
264        """
265        if content_type is None:
266            candidates = list(self._codecs)
267        else:
268            candidates = [codec for codec in self._codecs if codec.can_decode(content_type)]
269
270        if not candidates:
271            raise FatalError(
272                f"acemq: no codec here reads {content_type!r}; "
273                f"this one holds {self._describe()}"
274            )
275
276        refusals: list[str] = []
277        for codec in candidates:
278            try:
279                return codec.decode(body, content_type)
280            except Exception as failure:
281                # Collected rather than raised: a codec refusing says nothing
282                # about the next one, and the message that matters at the end is
283                # every reason together rather than whichever came last.
284                refusals.append(f"{type(codec).__name__}: {failure}")
285
286        raise FatalError(
287            "acemq: no codec could read this message. Tried " + "; ".join(refusals)
288        )
289
290    def can_decode(self, content_type: str | None) -> bool:
291        return any(codec.can_decode(content_type) for codec in self._codecs)
292
293    def _describe(self) -> str:
294        return ", ".join(codec.content_type for codec in self._codecs)
295
296    def __repr__(self) -> str:
297        return f"CompositeCodec({self._describe()})"
298
299
300_registry_lock = threading.Lock()
301_registry: dict[str, Callable[[], Codec]] = {}
302
303
304def register_codec(name: str, build: Callable[[], Codec]) -> None:
305    """Makes a codec available by name.
306
307    So that configuration can name a format without the code that reads the
308    configuration importing every format it might name. Registering the same
309    name twice replaces the first, which is what lets a test override a default.
310
311    :param name: what configuration will call it, such as ``"json"``
312    :param build: makes an instance
313    """
314    with _registry_lock:
315        _registry[name] = build
316
317
318def codec_by_name(name: str) -> Codec:
319    """Builds the codec registered under a name.
320
321    :param name: the registered name
322    :returns: a new instance
323    :raises KeyError: when nothing is registered under that name
324    """
325    with _registry_lock:
326        build = _registry.get(name)
327    if build is None:
328        raise KeyError(
329            f"acemq: no codec named {name!r} is registered; known: {', '.join(codec_names())}"
330        )
331    return build()
332
333
334def codec_names() -> list[str]:
335    """The registered codec names, sorted."""
336    with _registry_lock:
337        return sorted(_registry)
338
339
340register_codec("json", JsonCodec)
341register_codec("bytes", BytesCodec)
342register_codec("text", TextCodec)
343# The name .NET and Ruby register it under, so shared configuration works in all.
344register_codec("string", TextCodec)