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

1import base64 

2import datetime 

3import json 

4import pytz 

5import typing as t 

6from viur.core import db 

7 

8 

9class ViURJsonEncoder(json.JSONEncoder): 

10 """ 

11 Adds support for db.Key, db.Entity, datetime, bytes and and converts the provided obj 

12 into a special dict with JSON-serializable values. 

13 """ 

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

15 if isinstance(obj, bytes): 15 ↛ 16line 15 didn't jump to line 16 because the condition on line 15 was never true

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

17 elif isinstance(obj, datetime.datetime): 

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

19 elif isinstance(obj, datetime.timedelta): 

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

21 elif isinstance(obj, set): 

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

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

24 return tuple(obj) 

25 # cannot be tested in tests... 

26 elif isinstance(obj, db.Key): 

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

28 

29 return super().default(obj) 

30 

31 @staticmethod 

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

33 """ 

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

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

36 """ 

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

38 return { 

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

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

41 } 

42 elif isinstance(obj, dict): 

43 return { 

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

45 } 

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

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

48 

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

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

51 

52 return obj 

53 

54 

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

56 """ 

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

58 """ 

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

60 

61 

62def _decode_object_hook(obj: t.Any): 

63 """ 

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

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

66 """ 

67 if len(obj) == 1: 

68 if buf := obj.get(".__bytes__"): 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 return base64.b64decode(buf) 

70 elif date := obj.get(".__datetime__"): 

71 return datetime.datetime.fromisoformat(date) 

72 elif microseconds := obj.get(".__timedelta__"): 

73 return datetime.timedelta(microseconds=microseconds) 

74 elif key := obj.get(".__key__"): 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true

75 return db.Key.from_legacy_urlsafe(key) 

76 elif items := obj.get(".__set__"): 76 ↛ 84line 76 didn't jump to line 84 because the condition on line 76 was always true

77 return set(items) 

78 

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

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

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

82 return entity 

83 

84 return obj 

85 

86 

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

88 """ 

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

90 """ 

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