# # spyne - Copyright (C) Spyne contributors. # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public # License as published by the Free Software Foundation; either # version 2.1 of the License, or (at your option) any later version. # # This library is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 # """The ``spyne.protocol.xml`` module contains an xml-based protocol that serializes python objects to xml using Xml Schema conventions. Logs valid documents to ``'spyne.protocol.xml'`` and invalid documents to ``spyne.protocol.xml.invalid``. Use the usual ``logging.getLogger()`` and friends to configure how these get logged. Warning! You can get a lot of crap in the 'invalid' logger. You're not advised to turn it on for a production system. """ import logging logger = logging.getLogger('spyne.protocol.xml') logger_invalid = logging.getLogger('spyne.protocol.xml.invalid') from inspect import isgenerator from collections import defaultdict from lxml import etree from lxml import html from lxml.builder import E from lxml.etree import XMLSyntaxError from lxml.etree import XMLParser from spyne import BODY_STYLE_WRAPPED from spyne.util import Break, coroutine from spyne.util.six import text_type, string_types from spyne.util.cdict import cdict from spyne.util.etreeconv import etree_to_dict, dict_to_etree,\ root_dict_to_etree from spyne.const.xml import XSI, NS_SOAP11_ENC from spyne.error import Fault from spyne.error import ValidationError from spyne.const.ansi_color import LIGHT_GREEN from spyne.const.ansi_color import LIGHT_RED from spyne.const.ansi_color import END_COLOR from spyne.const.xml import NS_SOAP11_ENV from spyne.const.xml import PREFMAP, DEFAULT_NS from spyne.model import Any, ModelBase, Array, Iterable, ComplexModelBase, \ AnyHtml, AnyXml, AnyDict, Unicode, PushBase, File, ByteArray, XmlData, \ XmlAttribute from spyne.model.binary import BINARY_ENCODING_BASE64 from spyne.model.enum import EnumBase from spyne.protocol import ProtocolBase from spyne.util import six if six.PY2: STR_TYPES = (str, unicode) else: STR_TYPES = (str, bytes) NIL_ATTR = {XSI('nil'): 'true'} XSI_TYPE = XSI('type') def _append(parent, child_elt): if hasattr(parent, 'append'): parent.append(child_elt) else: parent.write(child_elt) def _gen_tagname(ns, name): if ns is not None: name = "{%s}%s" % (ns, name) return name class SchemaValidationError(Fault): """Raised when the input stream could not be validated by the Xml Schema.""" CODE = 'Client.SchemaValidationError' def __init__(self, faultstring): super(SchemaValidationError, self).__init__(self.CODE, faultstring) class SubXmlBase(ProtocolBase): def subserialize(self, ctx, cls, inst, parent, ns=None, name=None): return self.to_parent(ctx, cls, inst, parent, name) def to_parent(self, ctx, cls, inst, parent, ns, *args, **kwargs): """Serializes inst to an Element instance and appends it to the 'parent'. :param self: The protocol that will be used to serialize the given value. :param cls: The type of the value that's going to determine how to pack the given value. :param inst: The value to be set for the 'text' element of the newly created SubElement :param parent: The parent Element to which the new child will be appended. :param ns: The target namespace of the new SubElement, used with 'name' to set the tag. :param name: The tag name of the new SubElement, 'retval' by default. """ raise NotImplementedError() class XmlDocument(SubXmlBase): """The Xml input and output protocol, using the information from the Xml Schema generated by Spyne types. See the following material for more (much much more!) information. * http://www.w3.org/TR/xmlschema-0/ * http://www.w3.org/TR/xmlschema-1/ * http://www.w3.org/TR/xmlschema-2/ Receiving Xml from untrusted sources is a dodgy security dance as the Xml attack surface is /huge/. Spyne's ```lxml.etree.XMLParser``` instance has ```resolve_pis```, ```load_dtd```, ```resolve_entities```, ```dtd_validation```, ```huge_tree``` Defaults to ``False`` Having ```resolve_entities``` disabled will prevent the 'lxml' validation for documents with custom xml entities defined in the DTD. See the example in examples/xml/validation_error to play with the settings that work best for you. Please note that enabling ```resolve_entities``` is a security hazard that can lead to disclosure of sensitive information. See https://pypi.python.org/pypi/defusedxml for a pragmatic overview of Xml security in Python world. :param app: The owner application instance. :param validator: One of (None, 'soft', 'lxml', 'schema', ProtocolBase.SOFT_VALIDATION, XmlDocument.SCHEMA_VALIDATION). Both ``'lxml'`` and ``'schema'`` values are equivalent to ``XmlDocument.SCHEMA_VALIDATION``. Defaults to ``None``. :param replace_null_with_default: If ``False``, does not replace incoming explicit null values with denoted default values. This is against Xml Schema standard but consistent with other Spyne protocol implementations. Set this to False if you want cross-protocol compatibility. Defaults to ``True``. Relevant quote from xml schema primer (http://www.w3.org/TR/xmlschema-0/): .. When a value is present and is null The schema processor treats defaulted elements slightly differently. When an element is declared with a default value, the value of the element is whatever value appears as the element's content in the instance document; if the element appears without any content, the schema processor provides the element with a value equal to that of the default attribute. However, if the element does not appear in the instance document, the schema processor does not provide the element at all. In summary, the differences between element and attribute defaults can be stated as: Default attribute values apply when attributes are missing, and default element values apply when elements are empty. :param xml_declaration: Whether to add xml_declaration to the responses Defaults to ``True``. :param cleanup_namespaces: Whether to add clean up namespace declarations in the response document. Defaults to ``True``. :param encoding: The suggested string encoding for the returned xml documents. The transport can override this. Defaults to ``None``. :param pretty_print: When ``True``, returns the document in a pretty-printed format. Defaults to ``False``. :param parse_xsi_type: Set to ``False`` to disable parsing of ``xsi:type`` attribute, effectively disabling polymorphism. Defaults to ``True``. The following are passed straight to the ``XMLParser()`` instance from lxml. Docs are also plagiarized from the lxml documentation. Please note that some of the defaults are different to make parsing safer by default. :param attribute_defaults: read the DTD (if referenced by the document) and add the default attributes from it. Defaults to ``False`` :param dtd_validation: validate while parsing (if a DTD was referenced). Defaults to ``False`` :param load_dtd: load and parse the DTD while parsing (no validation is performed). Defaults to ``False``. :param no_network: prevent network access when looking up external documents. Defaults to ``True``. :param ns_clean: try to clean up redundant namespace declarations. Please note that this is for incoming documents. See ``cleanup_namespaces`` parameter for output documents. Defaults to ``False``. :param recover: try hard to parse through broken Xml. Defaults to ``False``. :param remove_blank_text: discard blank text nodes between tags, also known as ignorable whitespace. This is best used together with a DTD or schema (which tells data and noise apart), otherwise a heuristic will be applied. Defaults to ``False``. :param remove_pis: When ``True`` xml parser discards processing instructions. Defaults to ``True``. :param strip_cdata: replace CDATA sections by normal text content. Defaults to ``True`` :param resolve_entities: replace entities by their text value. Defaults to ``False``. :param huge_tree: disable security restrictions and support very deep trees and very long text content. (only affects libxml2 2.7+) Defaults to ``False``. :param compact: use compact storage for short text content. Defaults to ``True``. """ SCHEMA_VALIDATION = type("Schema", (object,), {}) mime_type = 'text/xml' default_binary_encoding = BINARY_ENCODING_BASE64 type = set(ProtocolBase.type) type.add('xml') soap_env = PREFMAP[NS_SOAP11_ENV] ns_soap_env = NS_SOAP11_ENV ns_soap_enc = NS_SOAP11_ENC def __init__(self, app=None, validator=None, replace_null_with_default=True, xml_declaration=True, cleanup_namespaces=True, encoding=None, pretty_print=False, attribute_defaults=False, dtd_validation=False, load_dtd=False, no_network=True, ns_clean=False, recover=False, remove_blank_text=False, remove_pis=True, strip_cdata=True, resolve_entities=False, huge_tree=False, compact=True, binary_encoding=None, parse_xsi_type=True, polymorphic=False, ): super(XmlDocument, self).__init__(app, validator, binary_encoding=binary_encoding) self.validation_schema = None self.xml_declaration = xml_declaration self.cleanup_namespaces = cleanup_namespaces self.replace_null_with_default = replace_null_with_default if encoding is None: self.encoding = 'UTF-8' else: self.encoding = encoding self.polymorphic = polymorphic self.pretty_print = pretty_print self.parse_xsi_type = parse_xsi_type self.serialization_handlers = cdict({ Any: self.any_to_parent, Fault: self.fault_to_parent, EnumBase: self.enum_to_parent, AnyXml: self.any_xml_to_parent, XmlData: self.xmldata_to_parent, AnyDict: self.any_dict_to_parent, AnyHtml: self.any_html_to_parent, ModelBase: self.modelbase_to_parent, ByteArray: self.byte_array_to_parent, ComplexModelBase: self.complex_to_parent, XmlAttribute: self.xmlattribute_to_parent, SchemaValidationError: self.schema_validation_error_to_parent, }) self.deserialization_handlers = cdict({ AnyHtml: self.html_from_element, AnyXml: self.xml_from_element, Any: self.xml_from_element, Array: self.array_from_element, Fault: self.fault_from_element, AnyDict: self.dict_from_element, EnumBase: self.enum_from_element, ModelBase: self.base_from_element, Unicode: self.unicode_from_element, Iterable: self.iterable_from_element, ByteArray: self.byte_array_from_element, ComplexModelBase: self.complex_from_element, }) self.parser_kwargs = dict( attribute_defaults=attribute_defaults, dtd_validation=dtd_validation, load_dtd=load_dtd, no_network=no_network, ns_clean=ns_clean, recover=recover, remove_blank_text=remove_blank_text, remove_comments=True, remove_pis=remove_pis, strip_cdata=strip_cdata, resolve_entities=resolve_entities, huge_tree=huge_tree, compact=compact, encoding=encoding, ) def set_validator(self, validator): if validator in ('lxml', 'schema') or \ validator is self.SCHEMA_VALIDATION: self.validate_document = self.__validate_lxml self.validator = self.SCHEMA_VALIDATION elif validator == 'soft' or validator is self.SOFT_VALIDATION: self.validator = self.SOFT_VALIDATION elif validator is None: pass else: raise ValueError(validator) self.validation_schema = None def validate_body(self, ctx, message): """Sets ctx.method_request_string and calls :func:`generate_contexts` for validation.""" assert message in (self.REQUEST, self.RESPONSE), message line_header = LIGHT_RED + "Error:" + END_COLOR try: self.validate_document(ctx.in_body_doc) if message is self.REQUEST: line_header = LIGHT_GREEN + "Method request string:" + END_COLOR else: line_header = LIGHT_RED + "Response:" + END_COLOR finally: if logger.level == logging.DEBUG: logger.debug("%s %s" % (line_header, ctx.method_request_string)) logger.debug(etree.tostring(ctx.in_document, pretty_print=True)) def set_app(self, value): ProtocolBase.set_app(self, value) self.validation_schema = None if self.validator is self.SCHEMA_VALIDATION and value is not None: xml_schema = self.app.interface.docs.xml_schema xml_schema.build_validation_schema() self.validation_schema = xml_schema.validation_schema def __validate_lxml(self, payload): ret = self.validation_schema.validate(payload) logger.debug("Validated ? %r" % ret) if ret == False: error_text = text_type(self.validation_schema.error_log.last_error) raise SchemaValidationError(error_text.encode('ascii', 'xmlcharrefreplace')) def create_in_document(self, ctx, charset=None): """Uses the iterable of string fragments in ``ctx.in_string`` to set ``ctx.in_document``.""" string = b''.join(ctx.in_string) try: try: ctx.in_document = etree.fromstring(string, parser=XMLParser(**self.parser_kwargs)) except ValueError: logger.debug('ValueError: Deserializing from unicode strings ' 'with encoding declaration is not supported by ' 'lxml.') ctx.in_document = etree.fromstring(string.decode(charset), self.parser) except XMLSyntaxError as e: logger_invalid.error("%r in string %r", e, string) raise Fault('Client.XMLSyntaxError', str(e)) def decompose_incoming_envelope(self, ctx, message): assert message in (self.REQUEST, self.RESPONSE) ctx.in_header_doc = None # If you need header support, you should use Soap ctx.in_body_doc = ctx.in_document ctx.method_request_string = ctx.in_body_doc.tag self.validate_body(ctx, message) def from_element(self, ctx, cls, element): cls_attrs = self.get_cls_attrs(cls) if bool(element.get(XSI('nil'))): if self.validator is self.SOFT_VALIDATION and not \ cls_attrs.nillable: raise ValidationError(None) if self.replace_null_with_default: return cls_attrs.default return None # if present, use the xsi:type="ns0:ObjectName" # attribute to instantiate subclass objects if self.parse_xsi_type: xsi_type = element.get(XSI_TYPE, None) if xsi_type is not None: if ":" in xsi_type: prefix, objtype = xsi_type.split(':', 1) else: prefix, objtype = None, xsi_type ns = element.nsmap.get(prefix) if ns is not None: classkey = "{%s}%s" % (ns, objtype) else: logger.error("xsi:type namespace prefix " "'%s' in '%s' not found in %r", ns, xsi_type, element.nsmap) raise ValidationError(xsi_type) newclass = ctx.app.interface.classes.get(classkey, None) if newclass is None: logger.error("xsi:type '%s' interpreted as class key '%s' " "is not recognized", xsi_type, classkey) raise ValidationError(xsi_type) cls = newclass logger.debug("xsi:type '%s' overrides %r to %r", xsi_type, cls, newclass) handler = self.deserialization_handlers[cls] return handler(ctx, cls, element) def to_parent(self, ctx, cls, inst, parent, ns, *args, **kwargs): cls, add_type = self.get_polymorphic_target(cls, inst) cls_attrs = self.get_cls_attrs(cls) subprot = cls_attrs.prot if subprot is not None and isinstance(subprot, SubXmlBase): return subprot.subserialize(ctx, cls, inst, parent, ns, *args, **kwargs) handler = self.serialization_handlers[cls] if inst is None: inst = cls_attrs.default if inst is None: return self.null_to_parent(ctx, cls, inst, parent, ns, *args, **kwargs) if cls_attrs.exc: return kwargs['add_type'] = add_type return handler(ctx, cls, inst, parent, ns, *args, **kwargs) def deserialize(self, ctx, message): """Takes a MethodContext instance and a string containing ONE root xml tag. Returns the corresponding native python object. Not meant to be overridden. """ assert message in (self.REQUEST, self.RESPONSE) self.event_manager.fire_event('before_deserialize', ctx) if ctx.descriptor is None: if ctx.in_error is None: raise Fault("Client", "Method %r not found." % ctx.method_request_string) else: raise ctx.in_error if message is self.REQUEST: body_class = ctx.descriptor.in_message elif message is self.RESPONSE: body_class = ctx.descriptor.out_message # decode method arguments if ctx.in_body_doc is None: ctx.in_object = [None] * len(body_class._type_info) else: ctx.in_object = self.from_element(ctx, body_class, ctx.in_body_doc) if logger.level == logging.DEBUG and message is self.REQUEST: line_header = '%sRequest%s' % (LIGHT_GREEN, END_COLOR) outdoc_str = None if ctx.out_document is not None: outdoc_str = etree.tostring(ctx.out_document, xml_declaration=self.xml_declaration, pretty_print=True) logger.debug("%s %s" % (line_header, outdoc_str)) self.event_manager.fire_event('after_deserialize', ctx) def serialize(self, ctx, message): """Uses ``ctx.out_object``, ``ctx.out_header`` or ``ctx.out_error`` to set ``ctx.out_body_doc``, ``ctx.out_header_doc`` and ``ctx.out_document`` as an ``lxml.etree._Element instance``. Not meant to be overridden. """ assert message in (self.REQUEST, self.RESPONSE) self.event_manager.fire_event('before_serialize', ctx) if ctx.out_error is not None: tmp_elt = etree.Element('punk') retval = self.to_parent(ctx, ctx.out_error.__class__, ctx.out_error, tmp_elt, self.app.interface.get_tns()) ctx.out_document = tmp_elt[0] else: if message is self.REQUEST: result_message_class = ctx.descriptor.in_message elif message is self.RESPONSE: result_message_class = ctx.descriptor.out_message # assign raw result to its wrapper, result_message if ctx.descriptor.body_style == BODY_STYLE_WRAPPED: result_inst = result_message_class() for i, (k, v) in enumerate( result_message_class._type_info.items()): attrs = self.get_cls_attrs(v) result_inst._safe_set(k, ctx.out_object[i], v, attrs) else: result_inst = ctx.out_object if ctx.out_stream is None: tmp_elt = etree.Element('punk') retval = self.to_parent(ctx, result_message_class, result_inst, tmp_elt, self.app.interface.get_tns()) ctx.out_document = tmp_elt[0] else: retval = self.incgen(ctx, result_message_class, result_inst, self.app.interface.get_tns()) if self.cleanup_namespaces and ctx.out_document is not None: etree.cleanup_namespaces(ctx.out_document) self.event_manager.fire_event('after_serialize', ctx) return retval def create_out_string(self, ctx, charset=None): """Sets an iterable of string fragments to ctx.out_string""" if charset is None: charset = self.encoding ctx.out_string = [etree.tostring(ctx.out_document, encoding=charset, pretty_print=self.pretty_print, xml_declaration=self.xml_declaration)] if logger.level == logging.DEBUG: logger.debug('%sResponse%s %s' % (LIGHT_RED, END_COLOR, etree.tostring(ctx.out_document, pretty_print=True, encoding='UTF-8'))) @coroutine def incgen(self, ctx, cls, inst, ns, name=None): if name is None: name = cls.get_type_name() with etree.xmlfile(ctx.out_stream) as xf: ret = self.to_parent(ctx, cls, inst, xf, ns, name) if isgenerator(ret): try: while True: y = (yield) # may throw Break ret.send(y) except Break: try: ret.throw(Break()) except StopIteration: pass if hasattr(ctx.out_stream, 'finish'): ctx.out_stream.finish() def _gen_tag(self, cls, ns, name, add_type=False, **_): if ns is not None: name = "{%s}%s" % (ns, name) retval = E(name) if add_type: retval.attrib[XSI_TYPE] = cls.get_type_name_ns(self.app.interface) return retval def byte_array_to_parent(self, ctx, cls, inst, parent, ns, name='retval', **kwargs): elt = self._gen_tag(cls, ns, name, **kwargs) elt.text = self.to_unicode(cls, inst, self.binary_encoding) _append(parent, elt) def modelbase_to_parent(self, ctx, cls, inst, parent, ns, name='retval', **kwargs): elt = self._gen_tag(cls, ns, name, **kwargs) elt.text = self.to_unicode(cls, inst) _append(parent, elt) def null_to_parent(self, ctx, cls, inst, parent, ns, name='retval', **kwargs): if issubclass(cls, XmlAttribute): return elif issubclass(cls, XmlData): parent.attrib.update(NIL_ATTR) else: elt = self._gen_tag(cls, ns, name, **kwargs) elt.attrib.update(NIL_ATTR) _append(parent, elt) def null_from_element(self, ctx, cls, element): return None def xmldata_to_parent(self, ctx, cls, inst, parent, ns, name, add_type=False, **_): cls_attrs = self.get_cls_attrs(cls) ns = cls._ns if ns is None: ns = cls_attrs.sub_ns name = _gen_tagname(ns, name) if add_type: parent.attrib[XSI_TYPE] = cls.get_type_name_ns(self.app.interface) cls.marshall(self, name, inst, parent) def xmlattribute_to_parent(self, ctx, cls, inst, parent, ns, name, **_): ns = cls._ns cls_attrs = self.get_cls_attrs(cls) if ns is None: ns = cls_attrs.sub_ns name = _gen_tagname(ns, name) if inst is not None: if issubclass(cls.type, (ByteArray, File)): parent.set(name, self.to_unicode(cls.type, inst, self.binary_encoding)) else: parent.set(name, self.to_unicode(cls.type, inst)) @coroutine def gen_members_parent(self, ctx, cls, inst, parent, tag_name, subelts, add_type): attrib = {} if add_type: tnn = cls.get_type_name_ns(self.app.interface) if tnn != None: attrib[XSI_TYPE] = tnn else: # this only happens on incomplete interface states for eg. # get_object_as_xml where the full init is not performed for # perf reasons attrib[XSI_TYPE] = cls.get_type_name() if isinstance(parent, etree._Element): elt = etree.SubElement(parent, tag_name, attrib=attrib) elt.extend(subelts) ret = self._get_members_etree(ctx, cls, inst, elt) if isgenerator(ret): try: while True: y = (yield) # may throw Break ret.send(y) except Break: try: ret.throw(Break()) except StopIteration: pass else: with parent.element(tag_name, attrib=attrib): for e in subelts: parent.write(e) ret = self._get_members_etree(ctx, cls, inst, parent) if isgenerator(ret): try: while True: y = (yield) ret.send(y) except Break: try: ret.throw(Break()) except StopIteration: pass @coroutine def _get_members_etree(self, ctx, cls, inst, parent): try: parent_cls = getattr(cls, '__extends__', None) if not (parent_cls is None): ret = self._get_members_etree(ctx, parent_cls, inst, parent) if ret is not None: try: while True: sv2 = (yield) # may throw Break ret.send(sv2) except Break: try: ret.throw(Break()) except StopIteration: pass for k, v in cls._type_info.items(): sub_cls_attrs = self.get_cls_attrs(v) if sub_cls_attrs.exc: continue try: subvalue = getattr(inst, k, None) except: # e.g. SqlAlchemy could throw NoSuchColumnError subvalue = None # This is a tight loop, so enable this only when necessary. # logger.debug("get %r(%r) from %r: %r" % (k, v, inst, subvalue)) sub_ns = v.Attributes.sub_ns if sub_ns is None: sub_ns = cls.get_namespace() sub_name = v.Attributes.sub_name if sub_name is None: sub_name = k mo = v.Attributes.max_occurs if subvalue is not None and mo > 1: if isinstance(subvalue, PushBase): while True: sv = (yield) ret = self.to_parent(ctx, v, sv, parent, sub_ns, sub_name) if ret is not None: try: while True: sv2 = (yield) # may throw Break ret.send(sv2) except Break: try: ret.throw(Break()) except StopIteration: pass else: for sv in subvalue: ret = self.to_parent(ctx, v, sv, parent, sub_ns, sub_name) if ret is not None: try: while True: sv2 = (yield) # may throw Break ret.send(sv2) except Break: try: ret.throw(Break()) except StopIteration: pass # Don't include empty values for # non-nillable optional attributes. elif subvalue is not None or v.Attributes.min_occurs > 0: ret = self.to_parent(ctx, v, subvalue, parent, sub_ns, sub_name) if ret is not None: try: while True: sv2 = (yield) ret.send(sv2) except Break as b: try: ret.throw(b) except StopIteration: pass except Break: pass def complex_to_parent(self, ctx, cls, inst, parent, ns, name=None, add_type=False, **_): cls_attrs = self.get_cls_attrs(cls) sub_name = cls_attrs.sub_name if sub_name is not None: name = sub_name if name is None: name = cls.get_type_name() sub_ns = cls_attrs.sub_ns if not sub_ns in (None, DEFAULT_NS): ns = sub_ns tag_name = _gen_tagname(ns, name) inst = cls.get_serialization_instance(inst) return self.gen_members_parent(ctx, cls, inst, parent, tag_name, [], add_type) def _fault_to_parent_impl(self, ctx, cls, inst, parent, ns, subelts, **_): tag_name = "{%s}Fault" % self.ns_soap_env # Accepting raw lxml objects as detail is DEPRECATED. It's also not # documented. It's kept for backwards-compatibility purposes. if isinstance(inst.detail, string_types + (etree._Element,)): _append(subelts, E('detail', inst.detail)) elif isinstance(inst.detail, dict): if len(inst.detail) > 0: _append(subelts, root_dict_to_etree({'detail':inst.detail})) elif inst.detail is None: pass else: raise TypeError('Fault detail Must be dict, got', type(inst.detail)) # add other nonstandard fault subelements with get_members_etree return self.gen_members_parent(ctx, cls, inst, parent, tag_name, subelts, add_type=False) def fault_to_parent(self, ctx, cls, inst, parent, ns, *args, **kwargs): subelts = [ E("faultcode", '%s:%s' % (self.soap_env, inst.faultcode)), E("faultstring", inst.faultstring), E("faultactor", inst.faultactor), ] return self._fault_to_parent_impl(ctx, cls, inst, parent, ns, subelts) def schema_validation_error_to_parent(self, ctx, cls, inst, parent, ns,**_): subelts = [ E("faultcode", '%s:%s' % (self.soap_env, inst.faultcode)), # HACK: Does anyone know a better way of injecting raw xml entities? E("faultstring", html.fromstring(inst.faultstring).text), E("faultactor", inst.faultactor), ] if inst.detail != None: _append(subelts, E('detail', inst.detail)) # add other nonstandard fault subelements with get_members_etree return self._fault_to_parent_impl(ctx, cls, inst, parent, ns, subelts) def enum_to_parent(self, ctx, cls, inst, parent, ns, name='retval', **kwargs): self.modelbase_to_parent(ctx, cls, str(inst), parent, ns, name, **kwargs) def any_xml_to_parent(self, ctx, cls, inst, parent, ns, name, **_): if isinstance(inst, STR_TYPES): inst = etree.fromstring(inst) _append(parent, E(_gen_tagname(ns, name), inst)) def any_to_parent(self, ctx, cls, inst, parent, ns, name, **_): _append(parent, E(_gen_tagname(ns, name), inst)) def any_html_to_parent(self, ctx, cls, inst, parent, ns, name, **_): if isinstance(inst, string_types) and len(inst) > 0: inst = html.fromstring(inst) _append(parent, E(_gen_tagname(ns, name), inst)) def any_dict_to_parent(self, ctx, cls, inst, parent, ns, name, **_): elt = E(_gen_tagname(ns, name)) dict_to_etree(inst, elt) _append(parent, elt) def complex_from_element(self, ctx, cls, elt): inst = cls.get_deserialization_instance(ctx) flat_type_info = cls.get_flat_type_info(cls) # this is for validating cls.Attributes.{min,max}_occurs frequencies = defaultdict(int) cls_attrs = self.get_cls_attrs(cls) if cls_attrs._xml_tag_body_as is not None: for xtba_key, xtba_type in cls_attrs._xml_tag_body_as: xtba_attrs = self.get_cls_attrs(xtba_type.type) if issubclass(xtba_type.type, (ByteArray, File)): value = self.from_unicode(xtba_type.type, elt.text, self.binary_encoding) else: value = self.from_unicode(xtba_type.type, elt.text) inst._safe_set(xtba_key, value, xtba_type.type, xtba_attrs) # parse input to set incoming data to related attributes. for c in elt: if isinstance(c, etree._Comment): continue key = c.tag.split('}', 1)[-1] frequencies[key] += 1 member = flat_type_info.get(key, None) if member is None: member, key = cls._type_info_alt.get(key, (None, key)) if member is None: member, key = cls._type_info_alt.get(c.tag, (None, key)) if member is None: continue member_attrs = self.get_cls_attrs(member) mo = member_attrs.max_occurs if mo > 1: value = getattr(inst, key, None) if value is None: value = [] value.append(self.from_element(ctx, member, c)) else: value = self.from_element(ctx, member, c) inst._safe_set(key, value, member, member_attrs) for key, value_str in c.attrib.items(): submember = flat_type_info.get(key, None) if submember is None: submember, key = cls._type_info_alt.get(key, (None, key)) if submember is None: continue submember_attrs = self.get_cls_attrs(submember) mo = submember_attrs.max_occurs if mo > 1: value = getattr(inst, key, None) if value is None: value = [] value.append(self.from_unicode(submember.type, value_str)) else: value = self.from_unicode(submember.type, value_str) inst._safe_set(key, value, submember.type, submember_attrs) for key, value_str in elt.attrib.items(): member = flat_type_info.get(key, None) if member is None: member, key = cls._type_info_alt.get(key, (None, key)) if member is None: continue if not issubclass(member, XmlAttribute): continue if issubclass(member.type, (ByteArray, File)): value = self.from_unicode(member.type, value_str, self.binary_encoding) else: value = self.from_unicode(member.type, value_str) member_attrs = self.get_cls_attrs(member.type) inst._safe_set(key, value, member.type, member_attrs) if self.validator is self.SOFT_VALIDATION: for key, c in flat_type_info.items(): val = frequencies.get(key, 0) attr = self.get_cls_attrs(c) if val < attr.min_occurs or val > attr.max_occurs: raise Fault('Client.ValidationError', '%r member does not ' 'respect frequency constraints.' % key) return inst def array_from_element(self, ctx, cls, element): retval = [ ] (serializer,) = cls._type_info.values() for child in element.getchildren(): retval.append(self.from_element(ctx, serializer, child)) return retval def iterable_from_element(self, ctx, cls, element): (serializer,) = cls._type_info.values() for child in element.getchildren(): yield self.from_element(ctx, serializer, child) def enum_from_element(self, ctx, cls, element): if self.validator is self.SOFT_VALIDATION and not ( cls.validate_string(cls, element.text)): raise ValidationError(element.text) return getattr(cls, element.text) def fault_from_element(self, ctx, cls, element): code = element.find('faultcode').text string = element.find('faultstring').text factor = element.find('faultactor') if factor is not None: factor = factor.text detail = element.find('detail') return cls(faultcode=code, faultstring=string, faultactor=factor, detail=detail) def xml_from_element(self, ctx, cls, element): children = element.getchildren() retval = None if children: retval = element.getchildren()[0] return retval def html_from_element(self, ctx, cls, element): children = element.getchildren() retval = None if len(children) == 1: retval = children[0] # this is actually a workaround to a case that should never exist -- # anyXml types should only have one child tag. elif len(children) > 1: retval = E.html(*children) return retval def dict_from_element(self, ctx, cls, element): children = element.getchildren() if children: return etree_to_dict(element) return None def unicode_from_element(self, ctx, cls, element): if self.validator is self.SOFT_VALIDATION and not ( cls.validate_string(cls, element.text)): raise ValidationError(element.text) s = element.text if s is None: s = '' retval = self.from_unicode(cls, s) if self.validator is self.SOFT_VALIDATION and not ( cls.validate_native(cls, retval)): raise ValidationError(retval) return retval def base_from_element(self, ctx, cls, element): if self.validator is self.SOFT_VALIDATION and not ( cls.validate_string(cls, element.text)): raise ValidationError(element.text) retval = self.from_unicode(cls, element.text) if self.validator is self.SOFT_VALIDATION and not ( cls.validate_native(cls, retval)): raise ValidationError(retval) return retval def byte_array_from_element(self, ctx, cls, element): if self.validator is self.SOFT_VALIDATION and not ( cls.validate_string(cls, element.text)): raise ValidationError(element.text) retval = self.from_unicode(cls, element.text, self.binary_encoding) if self.validator is self.SOFT_VALIDATION and not ( cls.validate_native(cls, retval)): raise ValidationError(retval) return retval