| |
|
| | __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 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() |
| |
|
| |
|
