Coverage for adrs/json_utils.py: 91%

1import json 

2from datetime import datetime, timedelta 

3from decimal import Decimal 

4from typing import Any, Callable 

5 

6 

7class JSONEncoder(json.JSONEncoder): 

8 """Custom JSON Encoder to handle additional types.""" 

9 

10 def default(self, o): 

11 if isinstance(o, timedelta): 

12 return {"_type": "timedelta", "seconds": o.total_seconds()} 

13 elif isinstance(o, datetime): 

14 return {"_type": "datetime", "isoformat": o.isoformat()} 

15 elif isinstance(o, Decimal): 

16 return {"_type": "decimal", "value": str(o)} 

17 return super().default(o) 

18 

19 

20class JSONDecoder(json.JSONDecoder): 

21 """Custom JSON Decoder to handle additional types.""" 

22 

23 def __init__(self, *args, **kwargs): 

24 json.JSONDecoder.__init__( 

25 self, object_hook=JSONDecoder.custom_object_hook, *args, **kwargs 

26 ) 

27 

28 @staticmethod 

29 def custom_object_hook(obj): 

30 if "_type" in obj: 

31 match obj["_type"]: 

32 case "timedelta": 

33 return timedelta(seconds=obj["seconds"]) 

34 case "datetime": 

35 return datetime.fromisoformat(obj["isoformat"]) 

36 case "decimal": 

37 return Decimal(obj["value"]) 

38 case _: 

39 raise ValueError(f"Unknown type: {obj['_type']}") 

40 return obj 

41 

42 

43def dumps( 

44 obj: Any, 

45 *, 

46 skipkeys: bool = False, 

47 ensure_ascii: bool = True, 

48 check_circular: bool = True, 

49 allow_nan: bool = True, 

50 cls: type[json.JSONEncoder] | None = JSONEncoder, 

51 indent: None | int | str = None, 

52 separators: tuple[str, str] | None = None, 

53 default: Callable[[Any], Any] | None = None, 

54 sort_keys: bool = False, 

55 **kwds: Any, 

56) -> Any: 

57 """Serialize obj to a JSON formatted str using the custom JSONEncoder.""" 

58 return json.dumps( 

59 obj, 

60 skipkeys=skipkeys, 

61 ensure_ascii=ensure_ascii, 

62 check_circular=check_circular, 

63 allow_nan=allow_nan, 

64 cls=cls, 

65 indent=indent, 

66 separators=separators, 

67 default=default, 

68 sort_keys=sort_keys, 

69 **kwds, 

70 ) 

71 

72 

73def loads( 

74 s: str | bytes | bytearray, 

75 *, 

76 cls: type[json.JSONDecoder] | None = JSONDecoder, 

77 object_hook: Callable[[dict[Any, Any]], Any] | None = None, 

78 parse_float: Callable[[str], Any] | None = None, 

79 parse_int: Callable[[str], Any] | None = None, 

80 parse_constant: Callable[[str], Any] | None = None, 

81 object_pairs_hook: Callable[[list[tuple[Any, Any]]], Any] | None = None, 

82 **kwds: Any, 

83) -> Any: 

84 """Deserialize s (a str, bytes or bytearray instance containing a JSON document) to a Python object using the custom JSONDecoder.""" 

85 return json.loads( 

86 s, 

87 cls=cls, 

88 object_hook=object_hook, 

89 parse_float=parse_float, 

90 parse_int=parse_int, 

91 parse_constant=parse_constant, 

92 object_pairs_hook=object_pairs_hook, 

93 **kwds, 

94 )