codekingpro/portable-devtools
115k
1 2from __future__ import annotations3 4# The following YAML grammar is LL(1) and is parsed by a recursive descent5# parser.6#7# stream ::= STREAM-START implicit_document? explicit_document*8# STREAM-END9# implicit_document ::= block_node DOCUMENT-END*10# explicit_document ::= DIRECTIVE* DOCUMENT-START block_node? DOCUMENT-END*11# block_node_or_indentless_sequence ::=12# ALIAS13# | properties (block_content |14# indentless_block_sequence)?15# | block_content16# | indentless_block_sequence17# block_node ::= ALIAS18# | properties block_content?19# | block_content20# flow_node ::= ALIAS21# | properties flow_content?22# | flow_content23# properties ::= TAG ANCHOR? | ANCHOR TAG?24# block_content ::= block_collection | flow_collection | SCALAR25# flow_content ::= flow_collection | SCALAR26# block_collection ::= block_sequence | block_mapping27# flow_collection ::= flow_sequence | flow_mapping28# block_sequence ::= BLOCK-SEQUENCE-START (BLOCK-ENTRY block_node?)*29# BLOCK-END30# indentless_sequence ::= (BLOCK-ENTRY block_node?)+31# block_mapping ::= BLOCK-MAPPING_START32# ((KEY block_node_or_indentless_sequence?)?33# (VALUE block_node_or_indentless_sequence?)?)*34# BLOCK-END35# flow_sequence ::= FLOW-SEQUENCE-START36# (flow_sequence_entry FLOW-ENTRY)*37# flow_sequence_entry?38# FLOW-SEQUENCE-END39# flow_sequence_entry ::= flow_node | KEY flow_node? (VALUE flow_node?)?40# flow_mapping ::= FLOW-MAPPING-START41# (flow_mapping_entry FLOW-ENTRY)*42# flow_mapping_entry?43# FLOW-MAPPING-END44# flow_mapping_entry ::= flow_node | KEY flow_node? (VALUE flow_node?)?45#46# FIRST sets:47#48# stream: { STREAM-START <}49# explicit_document: { DIRECTIVE DOCUMENT-START }50# implicit_document: FIRST(block_node)51# block_node: { ALIAS TAG ANCHOR SCALAR BLOCK-SEQUENCE-START52# BLOCK-MAPPING-START FLOW-SEQUENCE-START FLOW-MAPPING-START }53# flow_node: { ALIAS ANCHOR TAG SCALAR FLOW-SEQUENCE-START FLOW-MAPPING-START }54# block_content: { BLOCK-SEQUENCE-START BLOCK-MAPPING-START55# FLOW-SEQUENCE-START FLOW-MAPPING-START SCALAR }56# flow_content: { FLOW-SEQUENCE-START FLOW-MAPPING-START SCALAR }57# block_collection: { BLOCK-SEQUENCE-START BLOCK-MAPPING-START }58# flow_collection: { FLOW-SEQUENCE-START FLOW-MAPPING-START }59# block_sequence: { BLOCK-SEQUENCE-START }60# block_mapping: { BLOCK-MAPPING-START }61# block_node_or_indentless_sequence: { ALIAS ANCHOR TAG SCALAR62# BLOCK-SEQUENCE-START BLOCK-MAPPING-START FLOW-SEQUENCE-START63# FLOW-MAPPING-START BLOCK-ENTRY }64# indentless_sequence: { ENTRY }65# flow_collection: { FLOW-SEQUENCE-START FLOW-MAPPING-START }66# flow_sequence: { FLOW-SEQUENCE-START }67# flow_mapping: { FLOW-MAPPING-START }68# flow_sequence_entry: { ALIAS ANCHOR TAG SCALAR FLOW-SEQUENCE-START69# FLOW-MAPPING-START KEY }70# flow_mapping_entry: { ALIAS ANCHOR TAG SCALAR FLOW-SEQUENCE-START71# FLOW-MAPPING-START KEY }72 73# need to have full path with import, as pkg_resources tries to load parser.py in __init__.py74# only to not do anything with the package afterwards75# and for Jython too76 77 78from ruamel.yaml.error import MarkedYAMLError79from ruamel.yaml.tokens import * # NOQA80from ruamel.yaml.events import * # NOQA81from ruamel.yaml.scanner import Scanner, RoundTripScanner, ScannerError # NOQA82from ruamel.yaml.scanner import BlankLineComment83from ruamel.yaml.comments import C_PRE, C_POST, C_SPLIT_ON_FIRST_BLANK84from ruamel.yaml.compat import nprint, nprintf # NOQA85from ruamel.yaml.tag import Tag86 87if False: # MYPY88 from typing import Any, Dict, Optional, List, Optional # NOQA89 90__all__ = ['Parser', 'RoundTripParser', 'ParserError']91 92 93def xprintf(*args: Any, **kw: Any) -> Any:94 return nprintf(*args, **kw)95 pass96 97 98class ParserError(MarkedYAMLError):99 pass100 101 102class Parser:103 # Since writing a recursive-descendant parser is a straightforward task, we104 # do not give many comments here.105 106 DEFAULT_TAGS = {'!': '!', '!!': 'tag:yaml.org,2002:'}107 108 def __init__(self, loader: Any) -> None:109 self.loader = loader110 if self.loader is not None and getattr(self.loader, '_parser', None) is None:111 self.loader._parser = self112 self.reset_parser()113 114 def reset_parser(self) -> None:115 # Reset the state attributes (to clear self-references)116 self.current_event = self.last_event = None117 self.tag_handles: Dict[Any, Any] = {}118 self.states: List[Any] = []119 self.marks: List[Any] = []120 self.state: Any = self.parse_stream_start121 122 def dispose(self) -> None:123 self.reset_parser()124 125 @property126 def scanner(self) -> Any:127 if hasattr(self.loader, 'typ'):128 return self.loader.scanner129 return self.loader._scanner130 131 @property132 def resolver(self) -> Any:133 if hasattr(self.loader, 'typ'):134 return self.loader.resolver135 return self.loader._resolver136 137 def check_event(self, *choices: Any) -> bool:138 # Check the type of the next event.139 if self.current_event is None:140 if self.state:141 self.current_event = self.state()142 if self.current_event is not None:143 if not choices:144 return True145 for choice in choices:146 if isinstance(self.current_event, choice):147 return True148 return False149 150 def peek_event(self) -> Any:151 # Get the next event.152 if self.current_event is None:153 if self.state:154 self.current_event = self.state()155 return self.current_event156 157 def get_event(self) -> Any:158 # Get the next event and proceed further.159 if self.current_event is None:160 if self.state:161 self.current_event = self.state()162 # assert self.current_event is not None163 # if self.current_event.end_mark.line != self.peek_event().start_mark.line:164 # xprintf('get_event', repr(self.current_event), self.peek_event().start_mark.line)165 self.last_event = value = self.current_event166 self.current_event = None167 return value168 169 # stream ::= STREAM-START implicit_document? explicit_document*170 # STREAM-END171 # implicit_document ::= block_node DOCUMENT-END*172 # explicit_document ::= DIRECTIVE* DOCUMENT-START block_node? DOCUMENT-END*173 174 def parse_stream_start(self) -> Any:175 # Parse the stream start.176 token = self.scanner.get_token()177 self.move_token_comment(token)178 event = StreamStartEvent(token.start_mark, token.end_mark, encoding=token.encoding)179 180 # Prepare the next state.181 self.state = self.parse_implicit_document_start182 183 return event184 185 def parse_implicit_document_start(self) -> Any:186 # Parse an implicit document.187 if not self.scanner.check_token(DirectiveToken, DocumentStartToken, StreamEndToken):188 # don't need copy, as an implicit tag doesn't add tag_handles189 self.tag_handles = self.DEFAULT_TAGS190 token = self.scanner.peek_token()191 start_mark = end_mark = token.start_mark192 event = DocumentStartEvent(start_mark, end_mark, explicit=False)193 194 # Prepare the next state.195 self.states.append(self.parse_document_end)196 self.state = self.parse_block_node197 198 return event199 200 else:201 return self.parse_document_start()202 203 def parse_document_start(self) -> Any:204 # Parse any extra document end indicators.205 while self.scanner.check_token(DocumentEndToken):206 self.scanner.get_token()207 # Parse an explicit document.208 if not self.scanner.check_token(StreamEndToken):209 version, tags = self.process_directives()210 if not self.scanner.check_token(DocumentStartToken):211 raise ParserError(212 None,213 None,214 "expected '<document start>', "215 f'but found {self.scanner.peek_token().id,!r}',216 self.scanner.peek_token().start_mark,217 )218 token = self.scanner.get_token()219 start_mark = token.start_mark220 end_mark = token.end_mark221 # if self.loader is not None and \222 # end_mark.line != self.scanner.peek_token().start_mark.line:223 # self.loader.scalar_after_indicator = False224 event: Any = DocumentStartEvent(225 start_mark,226 end_mark,227 explicit=True,228 version=version,229 tags=tags,230 comment=token.comment,231 )232 self.states.append(self.parse_document_end)233 self.state = self.parse_document_content234 else:235 # Parse the end of the stream.236 token = self.scanner.get_token()237 event = StreamEndEvent(token.start_mark, token.end_mark, comment=token.comment)238 assert not self.states239 assert not self.marks240 self.state = None241 return event242 243 def parse_document_end(self) -> Any:244 # Parse the document end.245 token = self.scanner.peek_token()246 start_mark = end_mark = token.start_mark247 explicit = False248 if self.scanner.check_token(DocumentEndToken):249 token = self.scanner.get_token()250 # if token.end_mark.line != self.peek_event().start_mark.line:251 pt = self.scanner.peek_token()252 if not isinstance(pt, StreamEndToken) and (253 token.end_mark.line == pt.start_mark.line254 ):255 raise ParserError(256 None,257 None,258 'found non-comment content after document end marker, '259 f'{self.scanner.peek_token().id,!r}',260 self.scanner.peek_token().start_mark,261 )262 end_mark = token.end_mark263 explicit = True264 event = DocumentEndEvent(start_mark, end_mark, explicit=explicit)265 266 # Prepare the next state.267 if self.resolver.processing_version == (1, 1):268 self.state = self.parse_document_start269 else:270 if explicit:271 # found a document end marker, can be followed by implicit document272 self.state = self.parse_implicit_document_start273 else:274 self.state = self.parse_document_start275 276 return event277 278 def parse_document_content(self) -> Any:279 if self.scanner.check_token(280 DirectiveToken, DocumentStartToken, DocumentEndToken, StreamEndToken,281 ):282 event = self.process_empty_scalar(self.scanner.peek_token().start_mark)283 self.state = self.states.pop()284 return event285 else:286 return self.parse_block_node()287 288 def process_directives(self) -> Any:289 yaml_version = None290 self.tag_handles = {}291 while self.scanner.check_token(DirectiveToken):292 token = self.scanner.get_token()293 if token.name == 'YAML':294 if yaml_version is not None:295 raise ParserError(296 None, None, 'found duplicate YAML directive', token.start_mark,297 )298 major, minor = token.value299 if major != 1:300 raise ParserError(301 None,302 None,303 'found incompatible YAML document (version 1.* is required)',304 token.start_mark,305 )306 yaml_version = token.value307 elif token.name == 'TAG':308 handle, prefix = token.value309 if handle in self.tag_handles:310 raise ParserError(311 None, None, f'duplicate tag handle {handle!r}', token.start_mark,312 )313 self.tag_handles[handle] = prefix314 if bool(self.tag_handles):315 value: Any = (yaml_version, self.tag_handles.copy())316 else:317 value = yaml_version, None318 if self.loader is not None and hasattr(self.loader, 'tags'):319 # ToDo: this is used to keep a single loaded file from losing its version320 # info, but it affects following versions that have no explicit directive321 self.loader.version = yaml_version322 if self.loader.tags is None:323 self.loader.tags = {}324 for k in self.tag_handles:325 self.loader.tags[k] = self.tag_handles[k]326 self.loader.doc_infos[-1].tags.append((k, self.tag_handles[k]))327 for key in self.DEFAULT_TAGS:328 if key not in self.tag_handles:329 self.tag_handles[key] = self.DEFAULT_TAGS[key]330 return value331 332 # block_node_or_indentless_sequence ::= ALIAS333 # | properties (block_content | indentless_block_sequence)?334 # | block_content335 # | indentless_block_sequence336 # block_node ::= ALIAS337 # | properties block_content?338 # | block_content339 # flow_node ::= ALIAS340 # | properties flow_content?341 # | flow_content342 # properties ::= TAG ANCHOR? | ANCHOR TAG?343 # block_content ::= block_collection | flow_collection | SCALAR344 # flow_content ::= flow_collection | SCALAR345 # block_collection ::= block_sequence | block_mapping346 # flow_collection ::= flow_sequence | flow_mapping347 348 def parse_block_node(self) -> Any:349 return self.parse_node(block=True)350 351 def parse_flow_node(self) -> Any:352 return self.parse_node()353 354 def parse_block_node_or_indentless_sequence(self) -> Any:355 return self.parse_node(block=True, indentless_sequence=True)356 357 # def transform_tag(self, handle: Any, suffix: Any) -> Any:358 # return self.tag_handles[handle] + suffix359 360 def select_tag_transform(self, tag: Tag) -> None:361 if tag is None:362 return363 tag.select_transform(False)364 365 def parse_node(self, block: bool = False, indentless_sequence: bool = False) -> Any:366 if self.scanner.check_token(AliasToken):367 token = self.scanner.get_token()368 event: Any = AliasEvent(token.value, token.start_mark, token.end_mark)369 self.state = self.states.pop()370 return event371 372 anchor = None373 tag = None374 start_mark = end_mark = tag_mark = None375 if self.scanner.check_token(AnchorToken):376 token = self.scanner.get_token()377 self.move_token_comment(token)378 start_mark = token.start_mark379 end_mark = token.end_mark380 anchor = token.value381 if self.scanner.check_token(TagToken):382 token = self.scanner.get_token()383 tag_mark = token.start_mark384 end_mark = token.end_mark385 # tag = token.value386 tag = Tag(387 handle=token.value[0], suffix=token.value[1], handles=self.tag_handles,388 )389 elif self.scanner.check_token(TagToken):390 token = self.scanner.get_token()391 try:392 self.move_token_comment(token)393 except NotImplementedError:394 pass395 start_mark = tag_mark = token.start_mark396 end_mark = token.end_mark397 # tag = token.value398 tag = Tag(handle=token.value[0], suffix=token.value[1], handles=self.tag_handles)399 if self.scanner.check_token(AnchorToken):400 token = self.scanner.get_token()401 start_mark = tag_mark = token.start_mark402 end_mark = token.end_mark403 anchor = token.value404 if tag is not None:405 self.select_tag_transform(tag)406 if tag.check_handle():407 raise ParserError(408 'while parsing a node',409 start_mark,410 f'found undefined tag handle {tag.handle!r}',411 tag_mark,412 )413 if start_mark is None:414 start_mark = end_mark = self.scanner.peek_token().start_mark415 event = None416 implicit = tag is None or str(tag) == '!'417 if indentless_sequence and self.scanner.check_token(BlockEntryToken):418 comment = None419 pt = self.scanner.peek_token()420 if self.loader and self.loader.comment_handling is None:421 if pt.comment and pt.comment[0]:422 comment = [pt.comment[0], []]423 pt.comment[0] = None424 elif pt.comment and pt.comment[0] is None and pt.comment[1]:425 comment = [None, pt.comment[1]]426 pt.comment[1] = None427 elif self.loader:428 if pt.comment:429 comment = pt.comment430 end_mark = self.scanner.peek_token().end_mark431 event = SequenceStartEvent(432 anchor, tag, implicit, start_mark, end_mark, flow_style=False, comment=comment,433 )434 self.state = self.parse_indentless_sequence_entry435 return event436 437 if self.scanner.check_token(ScalarToken):438 token = self.scanner.get_token()439 # self.scanner.peek_token_same_line_comment(token)440 end_mark = token.end_mark441 if (token.plain and tag is None) or str(tag) == '!':442 dimplicit = (True, False)443 elif tag is None:444 dimplicit = (False, True)445 else:446 dimplicit = (False, False)447 event = ScalarEvent(448 anchor,449 tag,450 dimplicit,451 token.value,452 start_mark,453 end_mark,454 style=token.style,455 comment=token.comment,456 )457 self.state = self.states.pop()458 elif self.scanner.check_token(FlowSequenceStartToken):459 pt = self.scanner.peek_token()460 end_mark = pt.end_mark461 event = SequenceStartEvent(462 anchor,463 tag,464 implicit,465 start_mark,466 end_mark,467 flow_style=True,468 comment=pt.comment,469 )470 self.state = self.parse_flow_sequence_first_entry471 elif self.scanner.check_token(FlowMappingStartToken):472 pt = self.scanner.peek_token()473 end_mark = pt.end_mark474 event = MappingStartEvent(475 anchor,476 tag,477 implicit,478 start_mark,479 end_mark,480 flow_style=True,481 comment=pt.comment,482 )483 self.state = self.parse_flow_mapping_first_key484 elif block and self.scanner.check_token(BlockSequenceStartToken):485 end_mark = self.scanner.peek_token().start_mark486 # should inserting the comment be dependent on the487 # indentation?488 pt = self.scanner.peek_token()489 comment = pt.comment490 # nprint('pt0', type(pt))491 if comment is None or comment[1] is None:492 comment = pt.split_old_comment()493 # nprint('pt1', comment)494 event = SequenceStartEvent(495 anchor, tag, implicit, start_mark, end_mark, flow_style=False, comment=comment,496 )497 self.state = self.parse_block_sequence_first_entry498 elif block and self.scanner.check_token(BlockMappingStartToken):499 end_mark = self.scanner.peek_token().start_mark500 comment = self.scanner.peek_token().comment501 event = MappingStartEvent(502 anchor, tag, implicit, start_mark, end_mark, flow_style=False, comment=comment,503 )504 self.state = self.parse_block_mapping_first_key505 elif anchor is not None or tag is not None:506 # Empty scalars are allowed even if a tag or an anchor is507 # specified.508 event = ScalarEvent(anchor, tag, (implicit, False), "", start_mark, end_mark)509 self.state = self.states.pop()510 else:511 if block:512 node = 'block'513 else:514 node = 'flow'515 token = self.scanner.peek_token()516 raise ParserError(517 f'while parsing a {node!s} node',518 start_mark,519 f'expected the node content, but found {token.id!r}',520 token.start_mark,521 )522 return event523 524 # block_sequence ::= BLOCK-SEQUENCE-START (BLOCK-ENTRY block_node?)*525 # BLOCK-END526 527 def parse_block_sequence_first_entry(self) -> Any:528 token = self.scanner.get_token()529 # move any comment from start token530 # self.move_token_comment(token)531 self.marks.append(token.start_mark)532 return self.parse_block_sequence_entry()533 534 def parse_block_sequence_entry(self) -> Any:535 if self.scanner.check_token(BlockEntryToken):536 token = self.scanner.get_token()537 self.move_token_comment(token)538 if not self.scanner.check_token(BlockEntryToken, BlockEndToken):539 self.states.append(self.parse_block_sequence_entry)540 return self.parse_block_node()541 else:542 self.state = self.parse_block_sequence_entry543 return self.process_empty_scalar(token.end_mark)544 if not self.scanner.check_token(BlockEndToken):545 token = self.scanner.peek_token()546 raise ParserError(547 'while parsing a block collection',548 self.marks[-1],549 f'expected <block end>, but found {token.id!r}',550 token.start_mark,551 )552 token = self.scanner.get_token() # BlockEndToken553 event = SequenceEndEvent(token.start_mark, token.end_mark, comment=token.comment)554 self.state = self.states.pop()555 self.marks.pop()556 return event557 558 # indentless_sequence ::= (BLOCK-ENTRY block_node?)+559 560 # indentless_sequence?561 # sequence:562 # - entry563 # - nested564 565 def parse_indentless_sequence_entry(self) -> Any:566 if self.scanner.check_token(BlockEntryToken):567 token = self.scanner.get_token()568 self.move_token_comment(token)569 if not self.scanner.check_token(570 BlockEntryToken, KeyToken, ValueToken, BlockEndToken,571 ):572 self.states.append(self.parse_indentless_sequence_entry)573 return self.parse_block_node()574 else:575 self.state = self.parse_indentless_sequence_entry576 return self.process_empty_scalar(token.end_mark)577 token = self.scanner.peek_token()578 c = None579 if self.loader and self.loader.comment_handling is None:580 c = token.comment581 start_mark = token.start_mark582 else:583 start_mark = self.last_event.end_mark # type: ignore584 c = self.distribute_comment(token.comment, start_mark.line) # type: ignore585 event = SequenceEndEvent(start_mark, start_mark, comment=c)586 self.state = self.states.pop()587 return event588 589 # block_mapping ::= BLOCK-MAPPING_START590 # ((KEY block_node_or_indentless_sequence?)?591 # (VALUE block_node_or_indentless_sequence?)?)*592 # BLOCK-END593 594 def parse_block_mapping_first_key(self) -> Any:595 token = self.scanner.get_token()596 self.marks.append(token.start_mark)597 return self.parse_block_mapping_key()598 599 def parse_block_mapping_key(self) -> Any:600 if self.scanner.check_token(KeyToken):601 token = self.scanner.get_token()602 self.move_token_comment(token)603 if not self.scanner.check_token(KeyToken, ValueToken, BlockEndToken):604 self.states.append(self.parse_block_mapping_value)605 return self.parse_block_node_or_indentless_sequence()606 else:607 self.state = self.parse_block_mapping_value608 return self.process_empty_scalar(token.end_mark)609 if self.resolver.processing_version > (1, 1) and self.scanner.check_token(ValueToken):610 self.state = self.parse_block_mapping_value611 return self.process_empty_scalar(self.scanner.peek_token().start_mark)612 if not self.scanner.check_token(BlockEndToken):613 token = self.scanner.peek_token()614 raise ParserError(615 'while parsing a block mapping',616 self.marks[-1],617 f'expected <block end>, but found {token.id!r}',618 token.start_mark,619 )620 token = self.scanner.get_token()621 self.move_token_comment(token)622 event = MappingEndEvent(token.start_mark, token.end_mark, comment=token.comment)623 self.state = self.states.pop()624 self.marks.pop()625 return event626 627 def parse_block_mapping_value(self) -> Any:628 if self.scanner.check_token(ValueToken):629 token = self.scanner.get_token()630 # value token might have post comment move it to e.g. block631 if self.scanner.check_token(ValueToken):632 self.move_token_comment(token)633 else:634 if not self.scanner.check_token(KeyToken):635 self.move_token_comment(token, empty=True)636 # else: empty value for this key cannot move token.comment637 if not self.scanner.check_token(KeyToken, ValueToken, BlockEndToken):638 self.states.append(self.parse_block_mapping_key)639 return self.parse_block_node_or_indentless_sequence()640 else:641 self.state = self.parse_block_mapping_key642 comment = token.comment643 if comment is None:644 token = self.scanner.peek_token()645 comment = token.comment646 if comment:647 token._comment = [None, comment[1]]648 comment = [comment[0], None]649 return self.process_empty_scalar(token.end_mark, comment=comment)650 else:651 self.state = self.parse_block_mapping_key652 token = self.scanner.peek_token()653 return self.process_empty_scalar(token.start_mark)654 655 # flow_sequence ::= FLOW-SEQUENCE-START656 # (flow_sequence_entry FLOW-ENTRY)*657 # flow_sequence_entry?658 # FLOW-SEQUENCE-END659 # flow_sequence_entry ::= flow_node | KEY flow_node? (VALUE flow_node?)?660 #661 # Note that while production rules for both flow_sequence_entry and662 # flow_mapping_entry are equal, their interpretations are different.663 # For `flow_sequence_entry`, the part `KEY flow_node? (VALUE flow_node?)?`664 # generate an inline mapping (set syntax).665 666 def parse_flow_sequence_first_entry(self) -> Any:667 token = self.scanner.get_token()668 self.marks.append(token.start_mark)669 return self.parse_flow_sequence_entry(first=True)670 671 def parse_flow_sequence_entry(self, first: bool = False) -> Any:672 if not self.scanner.check_token(FlowSequenceEndToken):673 if not first:674 if self.scanner.check_token(FlowEntryToken):675 self.scanner.get_token()676 else:677 token = self.scanner.peek_token()678 raise ParserError(679 'while parsing a flow sequence',680 self.marks[-1],681 f"expected ',' or ']', but got {token.id!r}",682 token.start_mark,683 )684 685 if self.scanner.check_token(KeyToken):686 token = self.scanner.peek_token()687 event: Any = MappingStartEvent(688 None, None, True, token.start_mark, token.end_mark, flow_style=True,689 )690 self.state = self.parse_flow_sequence_entry_mapping_key691 return event692 elif not self.scanner.check_token(FlowSequenceEndToken):693 self.states.append(self.parse_flow_sequence_entry)694 return self.parse_flow_node()695 token = self.scanner.get_token()696 event = SequenceEndEvent(token.start_mark, token.end_mark, comment=token.comment)697 self.state = self.states.pop()698 self.marks.pop()699 return event700 701 def parse_flow_sequence_entry_mapping_key(self) -> Any:702 token = self.scanner.get_token()703 if not self.scanner.check_token(ValueToken, FlowEntryToken, FlowSequenceEndToken):704 self.states.append(self.parse_flow_sequence_entry_mapping_value)705 return self.parse_flow_node()706 else:707 self.state = self.parse_flow_sequence_entry_mapping_value708 return self.process_empty_scalar(token.end_mark)709 710 def parse_flow_sequence_entry_mapping_value(self) -> Any:711 if self.scanner.check_token(ValueToken):712 token = self.scanner.get_token()713 if not self.scanner.check_token(FlowEntryToken, FlowSequenceEndToken):714 self.states.append(self.parse_flow_sequence_entry_mapping_end)715 return self.parse_flow_node()716 else:717 self.state = self.parse_flow_sequence_entry_mapping_end718 return self.process_empty_scalar(token.end_mark)719 else:720 self.state = self.parse_flow_sequence_entry_mapping_end721 token = self.scanner.peek_token()722 return self.process_empty_scalar(token.start_mark)723 724 def parse_flow_sequence_entry_mapping_end(self) -> Any:725 self.state = self.parse_flow_sequence_entry726 token = self.scanner.peek_token()727 return MappingEndEvent(token.start_mark, token.start_mark)728 729 # flow_mapping ::= FLOW-MAPPING-START730 # (flow_mapping_entry FLOW-ENTRY)*731 # flow_mapping_entry?732 # FLOW-MAPPING-END733 # flow_mapping_entry ::= flow_node | KEY flow_node? (VALUE flow_node?)?734 735 def parse_flow_mapping_first_key(self) -> Any:736 token = self.scanner.get_token()737 self.marks.append(token.start_mark)738 return self.parse_flow_mapping_key(first=True)739 740 def parse_flow_mapping_key(self, first: Any = False) -> Any:741 if not self.scanner.check_token(FlowMappingEndToken):742 if not first:743 if self.scanner.check_token(FlowEntryToken):744 self.scanner.get_token()745 else:746 token = self.scanner.peek_token()747 raise ParserError(748 'while parsing a flow mapping',749 self.marks[-1],750 f"expected ',' or '}}', but got {token.id!r}",751 token.start_mark,752 )753 if self.scanner.check_token(KeyToken):754 token = self.scanner.get_token()755 if not self.scanner.check_token(756 ValueToken, FlowEntryToken, FlowMappingEndToken,757 ):758 self.states.append(self.parse_flow_mapping_value)759 return self.parse_flow_node()760 else:761 self.state = self.parse_flow_mapping_value762 return self.process_empty_scalar(token.end_mark)763 elif self.resolver.processing_version > (1, 1) and self.scanner.check_token(764 ValueToken,765 ):766 self.state = self.parse_flow_mapping_value767 return self.process_empty_scalar(self.scanner.peek_token().end_mark)768 elif not self.scanner.check_token(FlowMappingEndToken):769 self.states.append(self.parse_flow_mapping_empty_value)770 return self.parse_flow_node()771 token = self.scanner.get_token()772 event = MappingEndEvent(token.start_mark, token.end_mark, comment=token.comment)773 self.state = self.states.pop()774 self.marks.pop()775 return event776 777 def parse_flow_mapping_value(self) -> Any:778 if self.scanner.check_token(ValueToken):779 token = self.scanner.get_token()780 if not self.scanner.check_token(FlowEntryToken, FlowMappingEndToken):781 self.states.append(self.parse_flow_mapping_key)782 return self.parse_flow_node()783 else:784 self.state = self.parse_flow_mapping_key785 return self.process_empty_scalar(token.end_mark)786 else:787 self.state = self.parse_flow_mapping_key788 token = self.scanner.peek_token()789 return self.process_empty_scalar(token.start_mark)790 791 def parse_flow_mapping_empty_value(self) -> Any:792 self.state = self.parse_flow_mapping_key793 return self.process_empty_scalar(self.scanner.peek_token().start_mark)794 795 def process_empty_scalar(self, mark: Any, comment: Any = None) -> Any:796 return ScalarEvent(None, None, (True, False), "", mark, mark, comment=comment)797 798 def move_token_comment(799 self, token: Any, nt: Optional[Any] = None, empty: Optional[bool] = False,800 ) -> Any:801 pass802 803 804class RoundTripParser(Parser):805 """roundtrip is a safe loader, that wants to see the unmangled tag"""806 807 def select_tag_transform(self, tag: Tag) -> None:808 if tag is None:809 return810 tag.select_transform(True)811 812 def move_token_comment(813 self, token: Any, nt: Optional[Any] = None, empty: Optional[bool] = False,814 ) -> Any:815 token.move_old_comment(self.scanner.peek_token() if nt is None else nt, empty=empty)816 817 818class RoundTripParserSC(RoundTripParser):819 """roundtrip is a safe loader, that wants to see the unmangled tag"""820 821 # some of the differences are based on the superclass testing822 # if self.loader.comment_handling is not None823 824 def move_token_comment(825 self: Any, token: Any, nt: Any = None, empty: Optional[bool] = False,826 ) -> None:827 token.move_new_comment(self.scanner.peek_token() if nt is None else nt, empty=empty)828 829 def distribute_comment(self, comment: Any, line: Any) -> Any:830 # ToDo, look at indentation of the comment to determine attachment831 if comment is None:832 return None833 if not comment[0]:834 return None835 # if comment[0][0] != line + 1:836 # nprintf('>>>dcxxx', comment, line)837 assert comment[0][0] == line + 1838 # if comment[0] - line > 1:839 # return840 typ = self.loader.comment_handling & 0b11841 # nprintf('>>>dca', comment, line, typ)842 if typ == C_POST:843 return None844 if typ == C_PRE:845 c = [None, None, comment[0]]846 comment[0] = None847 return c848 # nprintf('>>>dcb', comment[0])849 for _idx, cmntidx in enumerate(comment[0]):850 # nprintf('>>>dcb', cmntidx)851 if isinstance(self.scanner.comments[cmntidx], BlankLineComment):852 break853 else:854 return None # no space found855 if _idx == 0:856 return None # first line was blank857 # nprintf('>>>dcc', idx)858 if typ == C_SPLIT_ON_FIRST_BLANK:859 c = [None, None, comment[0][:_idx]]860 comment[0] = comment[0][_idx:]861 return c862 raise NotImplementedError # reserved863