faust.utils.json

JSON utilities.

faust.utils.json.str_to_decimal(s: str, maxlen: int = 1000) → Union[decimal.Decimal, NoneType]

Convert string to Decimal.

Parameters:

• s (str) – Number to convert.
• maxlen (int) – Max length of string. Default is 100.

Raises:

ValueError – if length exceeds maximum length, or if value is not a valid number (e.g. Inf, NaN or sNaN).

Return type:

Optional[Decimal]

Returns:

Converted number.

Return type:

Decimal

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

default(o: Any, *, sequences: Tuple[type, ...] = (<class 'set'>, ), dates: Tuple[type, ...] = (<class 'datetime.date'>, <class 'datetime.time'>), value_delegate: Tuple[type, ...] = (<enum 'Enum'>, ), has_time: Tuple[type, ...] = (<class 'datetime.datetime'>, <class 'datetime.time'>), _isinstance: Callable = <built-in function isinstance>, _str: Callable = <class 'str'>, _list: Callable = <class 'list'>, textual: Tuple[type, ...] = (<class 'decimal.Decimal'>, <class 'uuid.UUID'>, <class 'bytes'>)) → Any

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):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

Return type:Any

faust.utils.json.dumps(obj: Any, cls: Type[faust.utils.json.JSONEncoder] = <class 'faust.utils.json.JSONEncoder'>, **kwargs) → str

Serialize to json. See json.dumps().

Return type:str

faust.utils.json.loads(s: str, **kwargs) → Any

Deserialize json string. See json.loads().

Return type:Any