faust.utils.json
¶
JSON utilities.
-
faust.utils.json.
str_to_decimal
(s: str, maxlen: int = 1000) → Optional[decimal.Decimal][source]¶ Convert string to
Decimal
.- Parameters
- Raises
ValueError – if length exceeds maximum length, or if value is not a valid number (e.g. Inf, NaN or sNaN).
- Return type
- 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)[source]¶ Faust customized
json.JSONEncoder
.Our version supports additional types like
UUID
, and importantly includes microsecond information in datetimes.-
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[source]¶ Implement this method in a subclass such that it returns a serializable object for
o
, or calls the base implementation (to raise aTypeError
).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
-
-
faust.utils.json.
dumps
(obj: Any, cls: Type[faust.utils.json.JSONEncoder] = <class 'faust.utils.json.JSONEncoder'>, **kwargs) → str[source]¶ Serialize to json. See
json.dumps()
.- Return type
-
faust.utils.json.
loads
(s: str, **kwargs) → Any[source]¶ Deserialize json string. See
json.loads()
.- Return type