Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/yaml/serializer.py: 18%

1 

2__all__ = ['Serializer', 'SerializerError'] 

3 

4from .error import YAMLError 

5from .events import * 

6from .nodes import * 

7 

8class SerializerError(YAMLError): 

9 pass 

10 

11class Serializer: 

12 

13 ANCHOR_TEMPLATE = 'id%03d' 

14 

15 def __init__(self, encoding=None, 

16 explicit_start=None, explicit_end=None, version=None, tags=None): 

17 self.use_encoding = encoding 

18 self.use_explicit_start = explicit_start 

19 self.use_explicit_end = explicit_end 

20 self.use_version = version 

21 self.use_tags = tags 

22 self.serialized_nodes = {} 

23 self.anchors = {} 

24 self.last_anchor_id = 0 

25 self.closed = None 

26 

27 def open(self): 

28 if self.closed is None: 

29 self.emit(StreamStartEvent(encoding=self.use_encoding)) 

30 self.closed = False 

31 elif self.closed: 

32 raise SerializerError("serializer is closed") 

33 else: 

34 raise SerializerError("serializer is already opened") 

35 

36 def close(self): 

37 if self.closed is None: 

38 raise SerializerError("serializer is not opened") 

39 elif not self.closed: 

40 self.emit(StreamEndEvent()) 

41 self.closed = True 

42 

43 #def __del__(self): 

44 # self.close() 

45 

46 def serialize(self, node): 

47 if self.closed is None: 

48 raise SerializerError("serializer is not opened") 

49 elif self.closed: 

50 raise SerializerError("serializer is closed") 

51 self.emit(DocumentStartEvent(explicit=self.use_explicit_start, 

52 version=self.use_version, tags=self.use_tags)) 

53 self.anchor_node(node) 

54 self.serialize_node(node, None, None) 

55 self.emit(DocumentEndEvent(explicit=self.use_explicit_end)) 

56 self.serialized_nodes = {} 

57 self.anchors = {} 

58 self.last_anchor_id = 0 

59 

60 def anchor_node(self, node): 

61 if node in self.anchors: 

62 if self.anchors[node] is None: 

63 self.anchors[node] = self.generate_anchor(node) 

64 else: 

65 self.anchors[node] = None 

66 if isinstance(node, SequenceNode): 

67 for item in node.value: 

68 self.anchor_node(item) 

69 elif isinstance(node, MappingNode): 

70 for key, value in node.value: 

71 self.anchor_node(key) 

72 self.anchor_node(value) 

73 

74 def generate_anchor(self, node): 

75 self.last_anchor_id += 1 

76 return self.ANCHOR_TEMPLATE % self.last_anchor_id 

77 

78 def serialize_node(self, node, parent, index): 

79 alias = self.anchors[node] 

80 if node in self.serialized_nodes: 

81 self.emit(AliasEvent(alias)) 

82 else: 

83 self.serialized_nodes[node] = True 

84 self.descend_resolver(parent, index) 

85 if isinstance(node, ScalarNode): 

86 detected_tag = self.resolve(ScalarNode, node.value, (True, False)) 

87 default_tag = self.resolve(ScalarNode, node.value, (False, True)) 

88 implicit = (node.tag == detected_tag), (node.tag == default_tag) 

89 self.emit(ScalarEvent(alias, node.tag, implicit, node.value, 

90 style=node.style)) 

91 elif isinstance(node, SequenceNode): 

92 implicit = (node.tag 

93 == self.resolve(SequenceNode, node.value, True)) 

94 self.emit(SequenceStartEvent(alias, node.tag, implicit, 

95 flow_style=node.flow_style)) 

96 index = 0 

97 for item in node.value: 

98 self.serialize_node(item, node, index) 

99 index += 1 

100 self.emit(SequenceEndEvent()) 

101 elif isinstance(node, MappingNode): 

102 implicit = (node.tag 

103 == self.resolve(MappingNode, node.value, True)) 

104 self.emit(MappingStartEvent(alias, node.tag, implicit, 

105 flow_style=node.flow_style)) 

106 for key, value in node.value: 

107 self.serialize_node(key, node, None) 

108 self.serialize_node(value, node, key) 

109 self.emit(MappingEndEvent()) 

110 self.ascend_resolver() 

111