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

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__": db.encodeKey(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 ↛ 39line 37 didn't jump to line 39 because the condition on line 37 was never true

38 # TODO: Handle SkeletonInstance as well? 

39 return { 

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

41 ".__key__": db.encodeKey(obj.key) if obj.key else None 

42 } 

43 elif isinstance(obj, dict): 

44 return { 

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

46 } 

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

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

49 

50 return obj 

51 

52 

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

54 """ 

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

56 """ 

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

58 

59 

60def _decode_object_hook(obj: t.Any): 

61 """ 

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

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

64 """ 

65 if len(obj) == 1: 

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

67 return base64.b64decode(buf) 

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

69 return datetime.datetime.fromisoformat(date) 

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

71 return datetime.timedelta(microseconds=microseconds) 

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

73 return db.Key.from_legacy_urlsafe(key) 

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

75 return set(items) 

76 

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

78 # TODO: Handle SkeletonInstance as well? 

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

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

81 return entity 

82 

83 return obj 

84 

85 

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

87 """ 

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

89 """ 

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