[docs]classXmlWriter(abc.ABC):""" A consistency wrapper for sax content handlers. - Implements a custom sax-like event api with separate start element/attribute events. - Buffers events until all content has been received or a child element is starting in order to build the current element's namespace context correctly. - Prepares values for serialization. :param config: Configuration instance :param output: Output text stream :param ns_map: User defined namespace prefix-URI map """__slots__=("config","output","ns_map","handler","in_tail","tail","attrs","ns_context","pending_tag","pending_prefixes",)def__init__(self,config:SerializerConfig,output:TextIO,ns_map:Dict,):self.config=configself.output=outputself.ns_map=ns_mapself.in_tail=Falseself.tail:Optional[str]=Noneself.attrs:Dict={}self.ns_context:List[Dict]=[]self.pending_tag:Optional[Tuple]=Noneself.pending_prefixes:List[List]=[]self.handler:ContentHandler
[docs]defwrite(self,events:Generator):""" Iterate over the generator events and feed the sax content handler with the information needed to generate the xml output. Example:: (XmlWriterEvent.START, "{http://www.w3.org/1999/xhtml}p"), (XmlWriterEvent.ATTR, "class", "paragraph"), (XmlWriterEvent.DATA, "Hello"), (XmlWriterEvent.END, "{http://www.w3.org/1999/xhtml}p"), :param events: Events generator """self.start_document()ifself.config.schema_location:self.add_attribute(QNames.XSI_SCHEMA_LOCATION,self.config.schema_location,check_pending=False,)ifself.config.no_namespace_schema_location:self.add_attribute(QNames.XSI_NO_NAMESPACE_SCHEMA_LOCATION,self.config.no_namespace_schema_location,check_pending=False,)forevent,*argsinevents:ifevent==XmlWriterEvent.START:self.start_tag(*args)elifevent==XmlWriterEvent.END:self.end_tag(*args)elifevent==XmlWriterEvent.ATTR:self.add_attribute(*args)elifevent==XmlWriterEvent.DATA:self.set_data(*args)else:raiseXmlWriterError(f"Unhandled event: `{event}`")self.handler.endDocument()
[docs]defstart_tag(self,qname:str):""" Start tag notification receiver. The receiver will flush the start of any pending element, create new namespaces context and queue the current tag for generation. :param qname: Tag qualified name """self.flush_start(False)self.ns_context.append(self.ns_map.copy())self.ns_map=self.ns_context[-1]self.pending_tag=split_qname(qname)self.add_namespace(self.pending_tag[0])
[docs]defadd_attribute(self,key:str,value:Any,check_pending:bool=True):""" Add attribute notification receiver. The receiver will convert the key to a namespace, name tuple and convert the value to string. Internally the converter will also generate any missing namespace prefixes. :param key: Attribute name :param value: Attribute value :param check_pending: Raise exception if not no element is pending start """ifnotself.pending_tagandcheck_pending:raiseXmlWriterError("Empty pending tag.")ifself.is_xsi_type(key,value):value=QName(value)name=split_qname(key)self.attrs[name]=self.encode_data(value)
[docs]defadd_namespace(self,uri:Optional[str]):""" Add the given uri to the current namespace context if the uri is valid and new. The prefix will be auto generated if it doesn't exist in the prefix-URI mappings. :param uri: Namespace uri """ifuriandnotprefix_exists(uri,self.ns_map):generate_prefix(uri,self.ns_map)
[docs]defset_data(self,data:Any):""" Set data notification receiver. The receiver will convert the data to string, flush any previous pending start element and send it to the handler for generation. If the text content of the tag has already been generated then treat the current data as element tail content and queue it to be generated when the tag ends. :param data: Element text or tail content """value=self.encode_data(data)self.flush_start(is_nil=valueisNone)ifvalue:ifnotself.in_tail:self.handler.characters(value)else:self.tail=valueself.in_tail=True
[docs]defend_tag(self,qname:str):""" End tag notification receiver. The receiver will flush if pending the start of the element, end the element, its tail content and its namespaces prefix mapping and current context. :param qname: Tag qualified name """self.flush_start(True)self.handler.endElementNS(split_qname(qname),None)ifself.tail:self.handler.characters(self.tail)self.tail=Noneself.in_tail=Falseself.ns_context.pop()ifself.ns_context:self.ns_map=self.ns_context[-1]forprefixinself.pending_prefixes.pop():self.handler.endPrefixMapping(prefix)
[docs]defflush_start(self,is_nil:bool=True):""" Flush start notification receiver. The receiver will pop the xsi:nil attribute if the element is not empty, prepare and send the namespaces prefix mappings and the element with its attributes to the content handler for generation. :param is_nil: If true add ``xsi:nil="true"`` to the element attributes """ifself.pending_tag:ifnotis_nil:self.attrs.pop(XSI_NIL,None)fornameinself.attrs.keys():self.add_namespace(name[0])self.reset_default_namespace()self.start_namespaces()self.handler.startElementNS(self.pending_tag,None,self.attrs)self.attrs={}self.in_tail=Falseself.pending_tag=None
[docs]defstart_namespaces(self):""" Send the new prefixes and namespaces added in the current context to the content handler. Save the list of prefixes to be removed at the end of the current pending tag. """prefixes=[]self.pending_prefixes.append(prefixes)try:parent_ns_map=self.ns_context[-2]exceptIndexError:parent_ns_map=EMPTY_MAPforprefix,uriinself.ns_map.items():ifparent_ns_map.get(prefix)!=uri:prefixes.append(prefix)self.handler.startPrefixMapping(prefix,uri)
[docs]defreset_default_namespace(self):"""Reset the default namespace if exists and the current pending tag is not qualified."""ifnotself.pending_tag[0]andNoneinself.ns_map:self.ns_map[None]=""
[docs]@classmethoddefis_xsi_type(cls,key:str,value:Any)->bool:""" Return whether the value is an xsi:type or not based on the given attribute name/value. :param key: Attribute name :param value: Attribute value """ifisinstance(value,str)andvalue.startswith("{"):returnkey==QNames.XSI_TYPEorDataType.from_qname(value)isnotNonereturnFalse
[docs]defencode_data(self,data:Any)->Optional[str]:"""Encode data for xml rendering."""ifdataisNoneorisinstance(data,str):returndataifisinstance(data,list)andnotdata:returnNonereturnconverter.serialize(data,ns_map=self.ns_map)