Create zeep.loader and move parser/utils/xsd.utils there

Fixes some duplicate coed

Author: mvantellingen

src/zeep/loader.py

@@ -0,0 +1,103 @@
+import os.path
+
+from defusedxml.lxml import fromstring
+from lxml import etree
+from six.moves.urllib.parse import urljoin, urlparse
+
+from zeep.exceptions import XMLSyntaxError
+
+
+class ImportResolver(etree.Resolver):
+    """Custom lxml resolve to use the transport object"""
+    def __init__(self, transport):
+        self.transport = transport
+
+    def resolve(self, url, pubid, context):
+        if urlparse(url).scheme in ('http', 'https'):
+            content = self.transport.load(url)
+            return self.resolve_string(content, context)
+
+
+def parse_xml(content, transport, base_url=None, recover=False):
+    """Parse an XML string and return the root Element.
+
+    :param content: The XML string
+    :type content: str
+    :param transport: The transport instance to load imported documents
+    :type transport: zeep.transports.Transport
+    :param base_url: The base url of the document, used to make relative
+      lookups absolute.
+    :type base_url: str
+    :param recover: boolean to indicate if the lxml recover mode should be
+      enabled. Recover mode tries to parse invalid XML as best as it can.
+    :type recover: boolean
+    :returns: The document root
+    :rtype: lxml.etree._Element
+
+    """
+    parser = etree.XMLParser(remove_comments=True, resolve_entities=False)
+    parser.resolvers.add(ImportResolver(transport))
+    try:
+        return fromstring(content, parser=parser, base_url=base_url)
+    except etree.XMLSyntaxError as exc:
+        raise XMLSyntaxError("Invalid XML content received (%s)" % exc.message)
+
+
+def load_external(url, transport, base_url=None):
+    """Load an external XML document.
+
+    :param url:
+    :param transport:
+    :param base_url:
+
+    """
+    if hasattr(url, 'read'):
+        content = url.read()
+    else:
+        if base_url:
+            url = absolute_location(url, base_url)
+        content = transport.load(url)
+    return parse_xml(content, transport, base_url)
+
+
+def absolute_location(location, base):
+    """Make an url absolute (if it is optional) via the passed base url.
+
+    :param location: The (relative) url
+    :type location: str
+    :param base: The base location
+    :type base: str
+    :returns: An absolute URL
+    :rtype: str
+
+    """
+    if location == base:
+        return location
+
+    if urlparse(location).scheme in ('http', 'https', 'file'):
+        return location
+
+    if base and urlparse(base).scheme in ('http', 'https', 'file'):
+        return urljoin(base, location)
+    else:
+        if os.path.isabs(location):
+            return location
+        if base:
+            return os.path.realpath(
+                os.path.join(os.path.dirname(base), location))
+    return location
+
+
+def is_relative_path(value):
+    """Check if the given value is a relative path
+
+    :param value: The value
+    :type value: str
+    :returns: Boolean indicating if the url is relative. If it is absolute then
+      False is returned.
+    :rtype: boolean
+
+    """
+    if urlparse(value).scheme in ('http', 'https', 'file'):
+        return False
+    return not os.path.isabs(value)

src/zeep/parser.py

@@ -5,8 +5,8 @@
 
 from zeep import ns, plugins, wsa
 from zeep.exceptions import Fault, TransportError, XMLSyntaxError
-from zeep.parser import parse_xml
-from zeep.utils import as_qname, qname_attr, get_media_type
+from zeep.loader import parse_xml
+from zeep.utils import as_qname, get_media_type, qname_attr
 from zeep.wsdl.attachments import MessagePack
 from zeep.wsdl.definitions import Binding, Operation
 from zeep.wsdl.messages import DocumentMessage, RpcMessage
@@ -145,7 +145,7 @@ def process_reply(self, client, operation, response):
             content = response.content
 
         try:
-            doc = parse_xml(content)
+            doc = parse_xml(content, self.transport)
         except XMLSyntaxError:
             raise TransportError(
                 u'Server returned HTTP status %d (%s)'

src/zeep/wsdl/bindings/soap.py

@@ -8,7 +8,7 @@
 import six
 from lxml import etree
 
-from zeep.parser import (
+from zeep.loader import (
     absolute_location, is_relative_path, load_external, parse_xml)
 from zeep.utils import findall_multiple_ns
 from zeep.wsdl import parse
@@ -65,7 +65,7 @@ def __init__(self, location, transport, base=None, strict=True):
             location=self.location,
             strict=self.strict)
 
-        document = self._load_content(location)
+        document = self._get_xml_document(location)
 
         root_definitions = Definition(self, document, self.location)
         root_definitions.resolve_imports()
@@ -117,16 +117,14 @@ def dump(self):
                     print('%s%s' % (' ' * 12, six.text_type(operation)))
                 print('')
 
-    def _load_content(self, location):
+    def _get_xml_document(self, location):
         """Load the XML content from the given location and return an
         lxml.Element object.
 
         :param location: The URL of the document to load
         :type location: string
 
         """
-        if hasattr(location, 'read'):
-            return parse_xml(location.read())
         return load_external(location, self.transport, self.location)
 
     def _add_definition(self, definition):
@@ -242,7 +240,7 @@ def parse_imports(self, doc):
             if key in self.wsdl._definitions:
                 self.imports[key] = self.wsdl._definitions[key]
             else:
-                document = self.wsdl._load_content(location)
+                document = self.wsdl._get_xml_document(location)
                 if etree.QName(document.tag).localname == 'schema':
                     self.types.add_documents([document], location)
                 else:

src/zeep/wsdl/wsdl.py

@@ -1,11 +1,6 @@
-from defusedxml.lxml import fromstring
-from lxml import etree
-
 from six.moves import range
-from six.moves.urllib.parse import urlparse
+
 from zeep import ns
-from zeep.exceptions import XMLSyntaxError
-from zeep.parser import absolute_location
 
 
 class NamePrefixGenerator(object):
@@ -32,34 +27,6 @@ def create_name(self, name):
             return name
 
 
-class ImportResolver(etree.Resolver):
-    """Custom lxml resolve to use the transport object"""
-    def __init__(self, transport):
-        self.transport = transport
-
-    def resolve(self, url, pubid, context):
-        if urlparse(url).scheme in ('http', 'https'):
-            content = self.transport.load(url)
-            return self.resolve_string(content, context)
-
-
-def parse_xml(content, transport, base_url=None):
-    parser = etree.XMLParser(remove_comments=True, resolve_entities=False)
-    parser.resolvers.add(ImportResolver(transport))
-    try:
-        return fromstring(content, parser=parser, base_url=base_url)
-    except etree.XMLSyntaxError as exc:
-        raise XMLSyntaxError("Invalid XML content received (%s)" % exc.message)
-
-
-def load_external(url, transport, base_url=None):
-    if base_url:
-        url = absolute_location(url, base_url)
-
-    response = transport.load(url)
-    return parse_xml(response, transport, base_url)
-
-
 def max_occurs_iter(max_occurs, items=None):
     assert max_occurs is not None
     generator = range(0, max_occurs if max_occurs != 'unbounded' else 2**31-1)

src/zeep/xsd/utils.py

@@ -6,12 +6,11 @@
 from lxml import etree
 
 from zeep.exceptions import XMLParseError
-from zeep.parser import absolute_location
+from zeep.loader import absolute_location, load_external
 from zeep.utils import as_qname, qname_attr
 from zeep.xsd import elements as xsd_elements
 from zeep.xsd import types as xsd_types
 from zeep.xsd.const import xsd_ns
-from zeep.xsd.utils import load_external
 
 logger = logging.getLogger(__name__)