FilesExpand file tree

 master

/

serializer.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

112 lines (98 loc) · 4.1 KB

 master

/

serializer.py

Copy path

Top

File metadata and controls

• Code
• Blame

112 lines (98 loc) · 4.1 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

# SPDX-License-Identifier: MIT

__all__ = ['Serializer', 'SerializerError']

from .error import YAMLError

from .events import *

from .nodes import *

class SerializerError(YAMLError):

pass

class Serializer:

ANCHOR_TEMPLATE = 'id%03d'

def __init__(self, encoding=None,

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

self.use_encoding = encoding

self.use_explicit_start = explicit_start

self.use_explicit_end = explicit_end

self.use_version = version

self.use_tags = tags

self.serialized_nodes = {}

self.anchors = {}

self.last_anchor_id = 0

self.closed = None

def open(self):

if self.closed is None:

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

self.closed = False

elif self.closed:

raise SerializerError("serializer is closed")

else:

raise SerializerError("serializer is already opened")

def close(self):

if self.closed is None:

raise SerializerError("serializer is not opened")

elif not self.closed:

self.emit(StreamEndEvent())

self.closed = True

#def __del__(self):

# self.close()

def serialize(self, node):

if self.closed is None:

raise SerializerError("serializer is not opened")

elif self.closed:

raise SerializerError("serializer is closed")

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

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

self.anchor_node(node)

self.serialize_node(node, None, None)

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

self.serialized_nodes = {}

self.anchors = {}

self.last_anchor_id = 0

def anchor_node(self, node):

if node in self.anchors:

if self.anchors[node] is None:

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

else:

self.anchors[node] = None

if isinstance(node, SequenceNode):

for item in node.value:

self.anchor_node(item)

elif isinstance(node, MappingNode):

for key, value in node.value:

self.anchor_node(key)

self.anchor_node(value)

def generate_anchor(self, node):

self.last_anchor_id += 1

return self.ANCHOR_TEMPLATE % self.last_anchor_id

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

alias = self.anchors[node]

if node in self.serialized_nodes:

self.emit(AliasEvent(alias))

else:

self.serialized_nodes[node] = True

self.descend_resolver(parent, index)

if isinstance(node, ScalarNode):

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

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

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

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

style=node.style))

elif isinstance(node, SequenceNode):

implicit = (node.tag

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

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

flow_style=node.flow_style))

index = 0

for item in node.value:

self.serialize_node(item, node, index)

index += 1

self.emit(SequenceEndEvent())

elif isinstance(node, MappingNode):

implicit = (node.tag

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

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

flow_style=node.flow_style))

for key, value in node.value:

self.serialize_node(key, node, None)

self.serialize_node(value, node, key)

self.emit(MappingEndEvent())

self.ascend_resolver()