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

JSON Utilities - kombu.utils.json

JSON Serialization Utilities.

class kombu.utils.json.JSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Kombu custom json encoder.


Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
        iterable = iter(o)
    except TypeError:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
kombu.utils.json.dumps(s, _dumps=<function dumps>, cls=<class 'kombu.utils.json.JSONEncoder'>, default_kwargs=None, **kwargs)[source]

Serialize object to json string.

kombu.utils.json.loads(s, _loads=<function loads>, decode_bytes=True, object_hook=<function object_hook>)[source]

Deserialize json from string.

kombu.utils.json.object_hook(o: dict)[source]

Hook function to perform custom deserialization.

kombu.utils.json.register_type(t: type[T], marker: str | None, encoder: Callable[[T], EncodedT], decoder: Callable[[EncodedT], T] = <function <lambda>>)[source]

Add support for serializing/deserializing native python type.

If marker is None, the encoding is a pure transformation and the result is not placed in an envelope, so decoder is unnecessary. Decoding must instead be handled outside this library.