Fix type errors.

examples/trafficvance_apikey.py

@@ -3,7 +3,7 @@
 API_KEY_TEST = 'YOUR_OWN_API_KEY'
 WSDL_TEST = 'https://apitest.trafficvance.com/?v3=system.wsdl'
 
-client = Client(WSDL)
+client = Client(WSDL_TEST)
 header = xsd.Element(
     '{WSDL_TEST}AuthenticateRequest',
     xsd.ComplexType([

mypy.ini

@@ -1,9 +1,9 @@
 [mypy]
+files = src/, benchmark/, examples/, tests/
 ignore_missing_imports = True
 implicit_reexport = False
 python_version = 3.6
 warn_unused_configs = True
 mypy_path = src
 warn_unreachable = True
-follow_imports = True
 follow_imports_for_stubs = True

src/zeep/__init__.py

@@ -1,7 +1,7 @@
-from zeep.client import AsyncClient, CachingClient, Client  # noqa
-from zeep.plugins import Plugin  # noqa
-from zeep.settings import Settings  # noqa
-from zeep.transports import Transport  # noqa
-from zeep.xsd.valueobjects import AnyObject  # noqa
+from zeep.client import AsyncClient as AsyncClient, CachingClient as CachingClient, Client as Client  # noqa
+from zeep.plugins import Plugin as Plugin  # noqa
+from zeep.settings import Settings as Settings  # noqa
+from zeep.transports import Transport as Transport  # noqa
+from zeep.xsd.valueobjects import AnyObject as AnyObject  # noqa
 
 __version__ = "4.1.0"

src/zeep/client.py

@@ -51,7 +51,7 @@ class Client:
 
     """
 
-    _default_transport = Transport
+    _default_transport = typing.Union[Transport, AsyncTransport]
 
     def __init__(
         self,

src/zeep/loader.py

@@ -68,7 +68,7 @@ def parse_xml(content: str, transport, base_url=None, settings=None):
         )
 
 
-def load_external(url: typing.IO, transport, base_url=None, settings=None):
+def load_external(url: typing.Union[typing.IO, str], transport, base_url=None, settings=None):
     """Load an external XML document.
 
     :param url:

src/zeep/wsdl/__init__.py

@@ -13,4 +13,4 @@
 
 
 """
-from zeep.wsdl.wsdl import Document  # noqa
+from zeep.wsdl.wsdl import Document as Document  # noqa

src/zeep/wsdl/attachments.py

@@ -5,10 +5,11 @@
 """
 
 import base64
+import sys
 
-try:
+if sys.version_info >= (3, 8):
     from functools import cached_property
-except ImportError:
+else:
     from cached_property import cached_property
 
 from requests.structures import CaseInsensitiveDict

src/zeep/wsdl/bindings/__init__.py

@@ -1,2 +1,2 @@
-from .http import HttpGetBinding, HttpPostBinding  # noqa
-from .soap import Soap11Binding, Soap12Binding  # noqa
+from .http import HttpGetBinding as HttpGetBinding, HttpPostBinding as HttpPostBinding  # noqa
+from .soap import Soap11Binding as Soap11Binding, Soap12Binding as Soap12Binding  # noqa

src/zeep/wsdl/definitions.py

@@ -25,6 +25,8 @@
 
 if typing.TYPE_CHECKING:
     from zeep.wsdl.wsdl import Definition
+else:
+    Definition = None
 
 MessagePart = namedtuple("MessagePart", ["element", "type"])
 
@@ -134,7 +136,7 @@ def __init__(self, wsdl, name, port_name):
         self.wsdl = wsdl
         self._operations = {}
 
-    def resolve(self, definitions: "Definition") -> None:
+    def resolve(self, definitions: Definition) -> None:
         self.port_type = definitions.get("port_types", self.port_name.text)
 
         for name, operation in list(self._operations.items()):

src/zeep/wsdl/wsdl.py

@@ -181,8 +181,8 @@ def __init__(self, wsdl, doc, location):
         self.types = wsdl.types
         self.port_types = {}
         self.messages = {}
-        self.bindings = {}  # type: typing.Dict[str, typing.Type[Binding]]
-        self.services = OrderedDict()  # type: typing.Dict[str, Service]
+        self.bindings: typing.Dict[str, Binding] = {}
+        self.services: typing.Dict[str, Service] = OrderedDict()
 
         self.imports = {}
         self._resolved_imports = False

src/zeep/wsse/__init__.py

@@ -1,3 +1,3 @@
-from .compose import Compose  # noqa
-from .signature import BinarySignature, MemorySignature, Signature  # noqa
-from .username import UsernameToken  # noqa
+from .compose import Compose as Compose  # noqa
+from .signature import BinarySignature as BinarySignature, MemorySignature as MemorySignature, Signature as Signature  # noqa
+from .username import UsernameToken as UsernameToken  # noqa

src/zeep/xsd/__init__.py

@@ -5,7 +5,7 @@
 """
 from zeep.xsd.const import Nil, SkipValue  # noqa
 from zeep.xsd.elements import *  # noqa
-from zeep.xsd.schema import Schema  # noqa
+from zeep.xsd.schema import Schema as Schema  # noqa
 from zeep.xsd.types import *  # noqa
 from zeep.xsd.types.builtins import *  # noqa
 from zeep.xsd.valueobjects import *  # noqa

src/zeep/xsd/elements/indicators.py

@@ -13,12 +13,13 @@
 """
 import copy
 import operator
+import sys
 import typing
 from collections import OrderedDict, defaultdict, deque
 
-try:
+if sys.version_info >= (3, 8):
     from functools import cached_property as threaded_cached_property
-except ImportError:
+else:
     from cached_property import threaded_cached_property
 
 from lxml import etree

src/zeep/xsd/types/any.py

@@ -1,16 +1,18 @@
 import logging
+import sys
 import typing
 
-try:
+if sys.version_info >= (3, 8):
     from functools import cached_property as threaded_cached_property
-except ImportError:
+else:
     from cached_property import threaded_cached_property
 
 from lxml import etree
 
 from zeep.utils import qname_attr
 from zeep.xsd.const import xsd_ns, xsi_ns
 from zeep.xsd.context import XmlParserContext
+from zeep.xsd.elements.base import Base
 from zeep.xsd.types.base import Type
 from zeep.xsd.valueobjects import AnyObject, CompoundValue
 
@@ -25,7 +27,7 @@
 
 class AnyType(Type):
     _default_qname = xsd_ns("anyType")
-    _element = None
+    _element: typing.Optional[Base] = None
 
     def __call__(self, value=None):
         return value or ""

src/zeep/xsd/types/complex.py

@@ -1,12 +1,14 @@
 import copy
 import logging
+import sys
 import typing
 from collections import OrderedDict, deque
 from itertools import chain
+from typing import Optional
 
-try:
+if sys.version_info >= (3, 8):
     from functools import cached_property as threaded_cached_property
-except ImportError:
+else:
     from cached_property import threaded_cached_property
 
 from lxml import etree
@@ -32,10 +34,14 @@
 if typing.TYPE_CHECKING:
     from zeep.xsd.schema import Schema
     from zeep.xsd.types.base import Type
+else:
+    Schema = Type = None
 
 logger = logging.getLogger(__name__)
 
 __all__ = ["ComplexType"]
+# Recursive alias
+_ObjectList = typing.List[typing.Union[CompoundValue, None, _ObjectList]]
 
 
 class ComplexType(AnyType):
@@ -166,10 +172,10 @@ def _array_type(self):
     def parse_xmlelement(
         self,
         xmlelement: etree._Element,
-        schema: "Schema" = None,
+        schema: Optional[Schema] = None,
         allow_none: bool = True,
         context: XmlParserContext = None,
-        schema_type: "Type" = None,
+        schema_type: Optional[Type] = None,
     ) -> typing.Optional[typing.Union[str, CompoundValue, typing.List[etree._Element]]]:
         """Consume matching xmlelements and call parse() on each
 
@@ -217,7 +223,7 @@ def parse_xmlelement(
 
             # Check if all children are consumed (parsed)
             if elements:
-                if schema.settings.strict:
+                if schema and schema.settings.strict:
                     raise XMLParseError("Unexpected element %r" % elements[0].tag)
                 else:
                     init_kwargs["_raw_elements"] = elements
@@ -333,8 +339,8 @@ def parse_kwargs(
         return {}
 
     def _create_object(
-        self, value: typing.Union[list, dict, CompoundValue], name: str
-    ) -> typing.Optional[CompoundValue]:
+        self, value: typing.Union[list, dict, CompoundValue, None], name: str
+    ) -> typing.Union[CompoundValue, None, _ObjectList]:
         """Return the value as a CompoundValue object
 
         :type value: str

tests/test_transports.py

@@ -32,7 +32,7 @@ def test_load():
 
 
 @pytest.mark.skipif(os.name != "nt", reason="test valid for windows platform only")
-def test_load_file():
+def test_load_file_windows():
     cache = stub(get=lambda url: None, add=lambda url, content: None)
     transport = transports.Transport(cache=cache)
     with patch("io.open", mock_open(read_data=b"x")) as m_open:
@@ -42,7 +42,7 @@ def test_load_file():
 
 
 @pytest.mark.skipif(os.name == "nt", reason="test valid for unix platform only")
-def test_load_file():
+def test_load_file_unix():
     cache = stub(get=lambda url: None, add=lambda url, content: None)
     transport = transports.Transport(cache=cache)
     with patch("io.open", mock_open(read_data=b"x")) as m_open:

tests/test_wsse_signature.py

@@ -9,7 +9,11 @@
 from zeep import ns, wsse
 from zeep.exceptions import SignatureVerificationFailed
 from zeep.wsse import signature
-from zeep.wsse.signature import xmlsec as xmlsec_installed
+
+try:
+    import xmlsec
+except ImportError:
+    xmlsec = None
 
 KEY_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), "cert_valid.pem")
 KEY_FILE_PW = os.path.join(
@@ -30,7 +34,7 @@
 skip_if_no_xmlsec = pytest.mark.skipif(
     sys.platform == "win32", reason="does not run on windows"
 ) and pytest.mark.skipif(
-    xmlsec_installed is None, reason="xmlsec library not installed"
+    xmlsec is None, reason="xmlsec library not installed"
 )
 
 
@@ -77,8 +81,8 @@ def test_sign_timestamp_if_present(
         KEY_FILE,
         KEY_FILE,
         None,
-        signature_method=getattr(xmlsec_installed.Transform, signature_method),
-        digest_method=getattr(xmlsec_installed.Transform, digest_method),
+        signature_method=getattr(xmlsec.Transform, signature_method),
+        digest_method=getattr(xmlsec.Transform, digest_method),
     )
     signature.verify_envelope(envelope, KEY_FILE)
     digests = envelope.xpath("//ds:DigestMethod", namespaces={"ds": ns.DS})
@@ -120,8 +124,8 @@ def test_sign(
         envelope,
         KEY_FILE,
         KEY_FILE,
-        signature_method=getattr(xmlsec_installed.Transform, signature_method),
-        digest_method=getattr(xmlsec_installed.Transform, digest_method),
+        signature_method=getattr(xmlsec.Transform, signature_method),
+        digest_method=getattr(xmlsec.Transform, digest_method),
     )
     signature.verify_envelope(envelope, KEY_FILE)
 
@@ -240,8 +244,8 @@ def test_signature_binary(
         KEY_FILE_PW,
         KEY_FILE_PW,
         "geheim",
-        signature_method=getattr(xmlsec_installed.Transform, signature_method),
-        digest_method=getattr(xmlsec_installed.Transform, digest_method),
+        signature_method=getattr(xmlsec.Transform, signature_method),
+        digest_method=getattr(xmlsec.Transform, digest_method),
     )
     envelope, headers = plugin.apply(envelope, {})
     plugin.verify(envelope)