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