Source code for xsdata.formats.dataclass.parsers.mixins
importabcfromtypingimportAnyfromtypingimportDictfromtypingimportListfromtypingimportOptionalfromtypingimportTuplefromtypingimportTypefromxsdata.exceptionsimportXmlHandlerErrorfromxsdata.formats.bindingsimportAbstractParserfromxsdata.formats.dataclass.parsers.configimportParserConfigfromxsdata.models.enumsimportEventTypeNoneStr=Optional[str]classPushParser(AbstractParser):""" A generic interface for event based content handlers like sax. :param config: Parser configuration. """config:ParserConfigns_map:Dict@abc.abstractmethoddefstart(self,clazz:Optional[Type],queue:List,objects:List,qname:str,attrs:Dict,ns_map:Dict,):"""Queue the next xml node for parsing."""@abc.abstractmethoddefend(self,queue:List,objects:List,qname:str,text:NoneStr,tail:NoneStr,)->bool:""" Parse the last xml node and bind any intermediate objects. :return: The result of the binding process. """defregister_namespace(self,prefix:NoneStr,uri:str):""" Add the given prefix-URI namespaces mapping if the prefix is new. :param prefix: Namespace prefix :param uri: Namespace uri """ifprefixnotinself.ns_map:self.ns_map[prefix]=uriclassXmlNode(abc.ABC):""" The xml node interface. The nodes are responsible to find and queue the child nodes when a new element starts and build the resulting object tree when the element ends. The parser needs to maintain a queue for these nodes and a list of all the intermediate object trees. """@abc.abstractmethoddefchild(self,qname:str,attrs:Dict,ns_map:Dict,position:int)->"XmlNode":""" Initialize the next child node to be queued, when a new xml element starts. This entry point is responsible to create the next node type with all the necessary information on how to bind the incoming input data. :param qname: Qualified name :param attrs: Attribute key-value map :param ns_map: Namespace prefix-URI map :param position: The current objects position, to mark future objects as children """@abc.abstractmethoddefbind(self,qname:str,text:NoneStr,tail:NoneStr,objects:List)->bool:""" Build the object tree for the ending element and return whether the result was successful or not. This entry point is called when an xml element ends and is responsible to parse the current element attributes/text, bind any children objects and initialize new object. :param qname: Qualified name :param text: Text content :param tail: Tail content :param objects: The list of intermediate parsed objects, eg [(qname, object)] """
[docs]classXmlHandler:""" Abstract content handler. :param parser: The parser instance to feed with events :param clazz: The target binding model, auto located if omitted. """__slots__=("parser","clazz","queue","objects")def__init__(self,parser:PushParser,clazz:Optional[Type]):self.parser=parserself.clazz=clazzself.queue:List=[]self.objects:List=[]
[docs]defparse(self,source:Any)->Any:"""Parse an XML document from a system identifier or an InputSource."""raiseNotImplementedError("This method must be implemented!")
[docs]defmerge_parent_namespaces(self,ns_map:Dict)->Dict:""" Merge and return the given prefix-URI map with the parent node. Register new prefixes with the parser. :param ns_map: Namespace prefix-URI map """ifself.queue:parent_ns_map=self.queue[-1].ns_mapifnotns_map:returnparent_ns_mapresult=parent_ns_map.copy()ifparent_ns_mapelse{}else:result={}forprefix,uriinns_map.items():self.parser.register_namespace(prefix,uri)result[prefix]=urireturnresult
classEventsHandler(XmlHandler):"""Sax content handler for pre-recorded events."""__slots__=("data_frames","flush_next")def__init__(self,parser:PushParser,clazz:Optional[Type]):super().__init__(parser,clazz)self.data_frames:List=[]self.flush_next:Optional[str]=Nonedefparse(self,source:List[Tuple])->Any:"""Forward the pre-recorded events to the main parser."""forevent,*argsinsource:ifevent==EventType.START:qname,attrs,ns_map=argsself.parser.start(self.clazz,self.queue,self.objects,qname,attrs,ns_map,)elifevent==EventType.END:qname,text,tail=argsself.parser.end(self.queue,self.objects,qname,text,tail)elifevent==EventType.START_NS:prefix,uri=argsself.parser.register_namespace(prefixorNone,uri)else:raiseXmlHandlerError(f"Unhandled event: `{event}`.")returnself.objects[-1][1]ifself.objectselseNone