Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/utils/json.py: 83%

1import base64 

2import datetime 

3import decimal 

4import json 

5import pytz 

6import typing as t 

7from viur.core import db 

8 

9 

10class ViURJsonEncoder(json.JSONEncoder): 

11 """ 

12 Adds support for db.Key, db.Entity, datetime, timedelta, bytes, set and Decimal and converts 

13 the provided obj into a special dict with JSON-serializable values. 

14 """ 

15 def default(self, obj: t.Any) -> t.Any: 

16 if isinstance(obj, bytes): 

17 return {".__bytes__": base64.b64encode(obj).decode("ASCII")} 

18 elif isinstance(obj, datetime.datetime): 

19 return {".__datetime__": obj.astimezone(pytz.UTC).isoformat()} 

20 elif isinstance(obj, datetime.timedelta): 

21 return {".__timedelta__": obj / datetime.timedelta(microseconds=1)} 

22 elif isinstance(obj, decimal.Decimal): 

23 # str() keeps the exact value (NumericBone(decimal=True)); float would round it 

24 return {".__decimal__": str(obj)} 

25 elif isinstance(obj, set): 

26 return {".__set__": list(obj)} 

27 elif hasattr(obj, "__iter__"): 27 ↛ 30line 27 didn't jump to line 30 because the condition on line 27 was always true

28 return tuple(obj) 

29 # cannot be tested in tests... 

30 elif isinstance(obj, db.Key): 

31 return {".__key__": str(obj)} 

32 

33 return super().default(obj) 

34 

35 @staticmethod 

36 def preprocess(obj: t.Any) -> t.Any: 

37 """ 

38 Needed to preprocess db.Entity as it subclasses dict. 

39 There is currently no other way to integrate with JSONEncoder. 

40 """ 

41 if isinstance(obj, db.Entity): 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 return { 

43 ".__entity__": ViURJsonEncoder.preprocess(dict(obj)), 

44 ".__key__": str(obj.key) if obj.key else None 

45 } 

46 elif isinstance(obj, dict): 

47 return { 

48 ViURJsonEncoder.preprocess(key): ViURJsonEncoder.preprocess(value) for key, value in obj.items() 

49 } 

50 elif isinstance(obj, (list, tuple)): 

51 return tuple(ViURJsonEncoder.preprocess(value) for value in obj) 

52 

53 elif hasattr(obj, "__class__") and obj.__class__.__name__ == "SkeletonInstance": # SkeletonInstance 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 return {bone_name: ViURJsonEncoder.preprocess(obj[bone_name]) for bone_name in obj} 

55 

56 return obj 

57 

58 

59def dumps(obj: t.Any, *, cls: ViURJsonEncoder = ViURJsonEncoder, **kwargs) -> str: 

60 """ 

61 Wrapper for json.dumps() which converts additional ViUR datatypes. 

62 """ 

63 return json.dumps(cls.preprocess(obj), cls=cls, **kwargs) 

64 

65 

66def _decode_object_hook(obj: t.Any): 

67 """ 

68 Inverse for _preprocess_json_object, which is an object-hook for json.loads. 

69 Check if the object matches a custom ViUR type and recreate it accordingly. 

70 """ 

71 # Membership checks, not truthiness: b"" encodes to "", timedelta(0) to 0.0 and 

72 # set() to [] -- all falsy, yet they must round-trip to their type, not to the marker dict. 

73 if len(obj) == 1: 

74 if ".__bytes__" in obj: 

75 return base64.b64decode(obj[".__bytes__"]) 

76 elif ".__datetime__" in obj: 

77 return datetime.datetime.fromisoformat(obj[".__datetime__"]) 

78 elif ".__timedelta__" in obj: 

79 return datetime.timedelta(microseconds=obj[".__timedelta__"]) 

80 elif ".__decimal__" in obj: 

81 return decimal.Decimal(obj[".__decimal__"]) 

82 elif ".__key__" in obj: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 return db.Key.from_legacy_urlsafe(obj[".__key__"]) 

84 elif ".__set__" in obj: 

85 return set(obj[".__set__"]) 

86 

87 elif len(obj) == 2 and all(k in obj for k in (".__entity__", ".__key__")): 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 entity = db.Entity(db.Key.from_legacy_urlsafe(obj[".__key__"]) if obj[".__key__"] else None) 

89 entity.update(obj[".__entity__"]) 

90 return entity 

91 

92 return obj 

93 

94 

95def loads(s: str, *, object_hook=_decode_object_hook, **kwargs) -> t.Any: 

96 """ 

97 Wrapper for json.loads() which recreates additional ViUR datatypes. 

98 """ 

99 return json.loads(s, object_hook=object_hook, **kwargs)