📄 xmlmarshaller.py
字号:
class NsElement(object): def __init__(self): self.nsMap = {} self.targetNS = None self.defaultNS = None self.prefix = None def __str__(self): if self.prefix == None: strVal = 'prefix = None; ' else: strVal = 'prefix = "%s"; ' % (self.prefix) if self.targetNS == None: strVal += 'targetNS = None; ' else: strVal += 'targetNS = "%s"; ' % (self.targetNS) if self.defaultNS == None: strVal += 'defaultNS = None; ' else: strVal += 'defaultNS = "%s"; ' % (self.defaultNS) if len(self.nsMap) == 0: strVal += 'nsMap = None; ' else: strVal += 'nsMap = {' for ik, iv in self.nsMap.iteritems(): strVal += '%s=%s; ' % (ik,iv) strVal += '}' return strVal def setKnownTypes(self, masterKnownTypes, masterKnownNamespaces, parentNSE): # if we're a nested element, extend our parent element's mapping if parentNSE != None: self.knownTypes = parentNSE.knownTypes.copy() # but if we have a different default namespace, replace the parent's default mappings if (self.defaultNS != None) and (parentNSE.defaultNS != self.defaultNS): newKT = self.knownTypes.copy() for tag in newKT: if tag.find(':') < 0: del self.knownTypes[tag] newMap = parentNSE.nsMap.copy() if self.nsMap != {}: for k, v in self.nsMap.iteritems(): newMap[k] = v self.nsMap = newMap else: self.knownTypes = {} reversedKNS = {} # TODO: instead of starting with the knownNamespaces, start with the "xmlms" mappings # for this element. Then we'd only process the namespaces and tags we need to. # But for now, this works. for long, short in masterKnownNamespaces.iteritems(): reversedKNS[short] = long mapLongs = self.nsMap.values() for tag, mapClass in masterKnownTypes.iteritems(): i = tag.rfind(':') if i >= 0: # e.g. "wsdl:description" knownTagShort = tag[:i] # "wsdl" knownTagName = tag[i+1:] # "description" knownTagLong = reversedKNS[knownTagShort] # e.g. "http://schemas.xmlsoap.org/wsdl" if (knownTagLong in mapLongs): for mShort, mLong in self.nsMap.iteritems(): if mLong == knownTagLong: actualShort = mShort # e.g. "ws" actualTag = '%s:%s' % (actualShort, knownTagName) self.knownTypes[actualTag] = mapClass break if self.defaultNS == knownTagLong: self.knownTypes[knownTagName] = mapClass else: # e.g. "ItemSearchRequest" self.knownTypes[tag] = mapClass def expandQName(self, eName, attrName, attrValue): bigValue = attrValue i = attrValue.rfind(':') if (i < 0): if self.defaultNS != None: bigValue = '%s:%s' % (self.defaultNS, attrValue) else: attrNS = attrValue[:i] attrNCName = attrValue[i+1:] for shortNs, longNs in self.nsMap.iteritems(): if shortNs == attrNS: bigValue = '%s:%s' % (longNs, attrNCName) break return bigValueclass XMLObjectFactory(xml.sax.ContentHandler): def __init__(self, knownTypes=None, knownNamespaces=None, xmlSource=None, createGenerics=False): self.rootelement = None if xmlSource == None: self.xmlSource = "unknown" else: self.xmlSource = xmlSource self.createGenerics = createGenerics self.skipper = False self.elementstack = [] self.nsstack = [] self.collectContent = None if (knownNamespaces == None): self.knownNamespaces = {} else: self.knownNamespaces = knownNamespaces self.reversedNamespaces = {} for longns, shortns in self.knownNamespaces.iteritems(): self.reversedNamespaces[shortns] = longns self.knownTypes = {} if (knownTypes != None): for tag, cls in knownTypes.iteritems(): i = tag.rfind(':') if i >= 0: shortns = tag[:i] tag = tag[i+1:] if not self.reversedNamespaces.has_key(shortns): errorString = 'Error unmarshalling XML document from source "%s": knownTypes specifies an unmapped short namespace "%s" for element "%s"' % (self.xmlSource, shortns, tag) raise UnmarshallerException(errorString) longns = self.reversedNamespaces[shortns] tag = '%s:%s' % (longns, tag) self.knownTypes[tag] = cls #printKnownTypes(self.knownTypes, 'Unmarshaller.XMLObjectFactory.__init__') xml.sax.handler.ContentHandler.__init__(self) def appendElementStack(self, newElement, newNS): self.elementstack.append(newElement) if (len(self.nsstack) > 0): oldNS = self.nsstack[-1] if newNS.defaultNS == None: newNS.defaultNS = oldNS.defaultNS if newNS.targetNS == None: newNS.targetNS = oldNS.targetNS if len(newNS.nsMap) == 0: newNS.nsMap = oldNS.nsMap elif len(oldNS.nsMap) > 0: map = oldNS.nsMap.copy() map.update(newNS.nsMap) newNS.nsMap = map self.nsstack.append(newNS) return newNS def popElementStack(self): element = self.elementstack.pop() nse = self.nsstack.pop() return element, nse ## ContentHandler methods def startElement(self, name, attrs):## print '[startElement] <%s>' % (name) if name == 'xs:annotation' or name == 'xsd:annotation': # should use namespace mapping here self.skipper = True self.appendElementStack(Element(name, attrs.copy()), NsElement()) if self.skipper: return if self.collectContent != None: strVal = '<%s' % (name) for aKey, aVal in attrs.items(): strVal += (' %s="%s"' % (aKey, aVal)) strVal += '>' self.collectContent.content += strVal xsname = name i = name.rfind(':') if i >= 0: nsname = name[:i] name = name[i+1:] else: nsname = None element = Element(name, attrs.copy(), xsname=xsname) # if the element has namespace attributes, process them and add them to our stack nse = NsElement() objtype = None for k in attrs.getNames(): if k.startswith('xmlns'): longNs = attrs[k] eLongNs = longNs + '/' if str(eLongNs) in asDict(self.knownNamespaces): longNs = eLongNs if k == 'xmlns': nse.defaultNS = longNs else: shortNs = k[6:] nse.nsMap[shortNs] = longNs elif k == 'targetNamespace': nse.targetNS = attrs.getValue(k) elif k == 'objtype': objtype = attrs.getValue(k) nse = self.appendElementStack(element, nse) if nsname != None: if nse.nsMap.has_key(nsname): longname = '%s:%s' % (nse.nsMap[nsname], name)## elif objtype == None:## errorString = 'Error unmarshalling XML document from source "%s": tag "%s" at line "%d", column "%d" has an undefined namespace' % (self.xmlSource, xsname, self._locator.getLineNumber(), self._locator.getColumnNumber())## raise UnmarshallerException(errorString) elif self.reversedNamespaces.has_key(nsname): longname = '%s:%s' % (self.reversedNamespaces[nsname], name) else: longname = xsname elif nse.defaultNS != None: longname = '%s:%s' % (nse.defaultNS, name) else: longname = name element.objtype = objtype element.objclass = self.knownTypes.get(longname) if element.objclass == None and len(self.knownNamespaces) == 0: # handles common case where tags are unqualified and knownTypes are too, but there's a defaultNS element.objclass = self.knownTypes.get(name) if (hasattr(element.objclass, "__xmlcontent__")): self.collectContent = element def characters(self, content):## print '[characters] "%s" (%s)' % (content, type(content)) if (content != None): if self.collectContent != None: self.collectContent.content += content else: self.elementstack[-1].content += content def endElement(self, name):## print "[endElement] </%s>" % name xsname = name i = name.rfind(':') if i >= 0: # Strip namespace prefixes for now until actually looking them up in xsd name = name[i+1:] if self.skipper: if xsname == "xs:annotation" or xsname == "xsd:annotation": # here too self.skipper = False self.popElementStack() return if self.collectContent != None: if xsname != self.collectContent.xsname: self.collectContent.content += ('</%s>' % (xsname)) self.popElementStack() return else: self.collectContent = None oldChildren = self.elementstack[-1].children element, nse = self.popElementStack() if ((len(self.elementstack) > 1) and (self.elementstack[-1].getobjtype() == "None")): parentElement = self.elementstack[-2] elif (len(self.elementstack) > 0): parentElement = self.elementstack[-1] objtype = element.getobjtype() if (objtype == "None"): return constructorarglist = [] if (len(element.content) > 0): strippedElementContent = element.content.strip() if (len(strippedElementContent) > 0): constructorarglist.append(element.content) # If the element requires an object, but none is known, use the GenericXMLObject class if ((element.objclass == None) and (element.attrs.get("objtype") == None) and ((len(element.attrs) > 0) or (len(element.children) > 0))): if self.createGenerics: element.objclass = GenericXMLObject obj = _objectfactory(objtype, constructorarglist, element.objclass) if element.objclass == GenericXMLObject: obj.setXMLAttributes(str(xsname), element.attrs, element.children, nse.nsMap, nse.defaultNS) complexType = getComplexType(obj) if (obj != None): if (hasattr(obj, "__xmlname__") and getattr(obj, "__xmlname__") == "sequence"): self.elementstack[-1].children = oldChildren return if (len(element.attrs) > 0) and not isinstance(obj, list): for attrname, attr in element.attrs.items(): if attrname == XMLNS or attrname.startswith(XMLNS_PREFIX): if attrname.startswith(XMLNS_PREFIX): ns = attrname[XMLNS_PREFIX_LENGTH:] else: ns = "" if complexType != None or element.objclass == GenericXMLObject: if not hasattr(obj, "__xmlnamespaces__"): obj.__xmlnamespaces__ = {ns:attr} elif ns not in obj.__xmlnamespaces__: if (hasattr(obj.__class__, "__xmlnamespaces__") and (obj.__xmlnamespaces__ is obj.__class__.__xmlnamespaces__)): obj.__xmlnamespaces__ = dict(obj.__xmlnamespaces__) obj.__xmlnamespaces__[ns] = attr elif not attrname == "objtype": if attrname.find(":") > -1: # Strip namespace prefixes for now until actually looking them up in xsd attrname = attrname[attrname.find(":") + 1:] if (complexType != None): xsdElement = complexType.findElement(attrname) if (xsdElement != None): type = xsdElement.type if (type != None): if (type == TYPE_QNAME): attr = nse.expandQName(name, attrname, attr) type = xsdToLangType(type) ### ToDO remove maxOccurs hack after bug 177 is fixed if attrname == "maxOccurs" and attr == "unbounded": attr = "-1" try: attr = _objectfactory(type, attr) except Exception, exceptData: errorString = 'Error unmarshalling attribute "%s" at line %d, column %d in XML document from source "%s": %s' % (attrname, self._locator.getLineNumber(), self._locator.getColumnNumber(), self.xmlSource, str(exceptData)) raise UnmarshallerException(errorString) try: setattrignorecase(obj, _toAttrName(obj, attrname), attr) except AttributeError: errorString = 'Error setting value of attribute "%s" at line %d, column %d in XML document from source "%s": object type of XML element "%s" is not specified or known' % (attrname, self._locator.getLineNumber(), self._locator.getColumnNumber(), self.xmlSource, name) raise UnmarshallerException(errorString)## obj.__dict__[_toAttrName(obj, attrname)] = attr # stuff any child attributes meant to be in a sequence via the __xmlflattensequence__ flattenDict = {} if hasattr(obj, "__xmlflattensequence__"): flatten = obj.__xmlflattensequence__ if (isinstance(flatten, dict)): for sequencename, xmlnametuple in flatten.items(): if (xmlnametuple == None): flattenDict[sequencename] = sequencename
⌨️ 快捷键说明
复制代码
Ctrl + C
搜索代码
Ctrl + F
全屏模式
F11
切换主题
Ctrl + Shift + D
显示快捷键
?
增大字号
Ctrl + =
减小字号
Ctrl + -