This document describes the current stable version of Kombu (5.3). For development docs, go here.

Source code for kombu.utils.json

"""JSON Serialization Utilities."""

from __future__ import annotations

import base64
import json
import uuid
from datetime import date, datetime, time
from decimal import Decimal
from typing import Any, Callable, TypeVar

textual_types = ()

    from django.utils.functional import Promise

    textual_types += (Promise,)
except ImportError:

[docs] class JSONEncoder(json.JSONEncoder): """Kombu custom json encoder."""
[docs] def default(self, o): reducer = getattr(o, "__json__", None) if reducer is not None: return reducer() if isinstance(o, textual_types): return str(o) for t, (marker, encoder) in _encoders.items(): if isinstance(o, t): return _as(marker, encoder(o)) # Bytes is slightly trickier, so we cannot put them directly # into _encoders, because we use two formats: bytes, and base64. if isinstance(o, bytes): try: return _as("bytes", o.decode("utf-8")) except UnicodeDecodeError: return _as("base64", base64.b64encode(o).decode("utf-8")) return super().default(o)
def _as(t: str, v: Any): return {"__type__": t, "__value__": v}
[docs] def dumps( s, _dumps=json.dumps, cls=JSONEncoder, default_kwargs=None, **kwargs ): """Serialize object to json string.""" default_kwargs = default_kwargs or {} return _dumps(s, cls=cls, **dict(default_kwargs, **kwargs))
[docs] def object_hook(o: dict): """Hook function to perform custom deserialization.""" if o.keys() == {"__type__", "__value__"}: decoder = _decoders.get(o["__type__"]) if decoder: return decoder(o["__value__"]) else: raise ValueError("Unsupported type", type, o) else: return o
[docs] def loads(s, _loads=json.loads, decode_bytes=True, object_hook=object_hook): """Deserialize json from string.""" # None of the json implementations supports decoding from # a buffer/memoryview, or even reading from a stream # (load is just loads( # but this is Python, we love copying strings, preferably many times # over. Note that pickle does support buffer/memoryview # </rant> if isinstance(s, memoryview): s = s.tobytes().decode("utf-8") elif isinstance(s, bytearray): s = s.decode("utf-8") elif decode_bytes and isinstance(s, bytes): s = s.decode("utf-8") return _loads(s, object_hook=object_hook)
DecoderT = EncoderT = Callable[[Any], Any] T = TypeVar("T") EncodedT = TypeVar("EncodedT")
[docs] def register_type( t: type[T], marker: str, encoder: Callable[[T], EncodedT], decoder: Callable[[EncodedT], T], ): """Add support for serializing/deserializing native python type.""" _encoders[t] = (marker, encoder) _decoders[marker] = decoder
_encoders: dict[type, tuple[str, EncoderT]] = {} _decoders: dict[str, DecoderT] = { "bytes": lambda o: o.encode("utf-8"), "base64": lambda o: base64.b64decode(o.encode("utf-8")), } # NOTE: datetime should be registered before date, # because datetime is also instance of date. register_type(datetime, "datetime", datetime.isoformat, datetime.fromisoformat) register_type( date, "date", lambda o: o.isoformat(), lambda o: datetime.fromisoformat(o).date(), ) register_type(time, "time", lambda o: o.isoformat(), time.fromisoformat) register_type(Decimal, "decimal", str, Decimal) register_type( uuid.UUID, "uuid", lambda o: {"hex": o.hex}, lambda o: uuid.UUID(**o), )