Fix WSDL files being loaded multiple times and blocking the event loop (#21)

Author: bdraco

onvif/client.py

@@ -6,17 +6,19 @@
 import logging
 import os.path
 import ssl
-from typing import Any, Dict, Optional, Tuple
+from typing import IO, Any, Dict, Optional, Tuple, Union
 
 import httpx
 from httpx import AsyncClient, BasicAuth, DigestAuth
 from zeep.cache import SqliteCache
 from zeep.client import AsyncClient as BaseZeepAsyncClient, Settings
 from zeep.exceptions import Fault
 import zeep.helpers
+from zeep.loader import parse_xml
 from zeep.proxy import AsyncServiceProxy
 from zeep.transports import AsyncTransport
 from zeep.wsa import WsAddressingPlugin
+from zeep.wsdl import Document
 from zeep.wsse.username import UsernameToken
 
 from onvif.definition import SERVICES
@@ -26,6 +28,10 @@
 logging.basicConfig(level=logging.INFO)
 logging.getLogger("zeep.client").setLevel(logging.CRITICAL)
 
+_DEFAULT_SETTINGS = Settings()
+_DEFAULT_SETTINGS.strict = False
+_DEFAULT_SETTINGS.xml_huge_tree = True
+
 
 def create_no_verify_ssl_context() -> ssl.SSLContext:
     """Return an SSL context that does not verify the server certificate.
@@ -85,6 +91,41 @@ def apply(self, envelope, headers):
         return result
 
 
+class AsyncSafeTransport:
+    """A transport that blocks all I/O for zeep."""
+
+    def load(self, *args: Any, **kwargs: Any) -> None:
+        """Load the given XML document.
+
+        This should never be called, but we want to raise
+        an error if it is so we know we're doing something wrong
+        and do not accidentally block the event loop.
+        """
+        raise RuntimeError("Loading is not supported in async mode")
+
+
+_ASYNC_TRANSPORT = AsyncSafeTransport()
+
+
+@lru_cache(maxsize=128)
+def _cached_parse_xml(path: str) -> Any:
+    """Load external XML document from disk."""
+    with open(os.path.expanduser(path), "rb") as fh:
+        return parse_xml(fh.read(), _ASYNC_TRANSPORT, settings=_DEFAULT_SETTINGS)
+
+
+class DocumentWithCache(Document):
+    """A WSDL document that supports caching."""
+
+    def _get_xml_document(self, url: Union[IO, str]) -> Any:
+        """Load external XML document from a file-like object or URL."""
+        if _path_isfile(url):
+            return _cached_parse_xml(url)
+        raise RuntimeError(
+            f"Cannot fetch {url} in async mode because it would block the event loop"
+        )
+
+
 class ZeepAsyncClient(BaseZeepAsyncClient):
     """Overwrite create_service method to be async."""
 
@@ -174,21 +215,18 @@ def __init__(
                 client=client, wsdl_client=wsdl_client, cache=SqliteCache()
             )
         )
-        settings = Settings()
-        settings.strict = False
-        settings.xml_huge_tree = True
+        settings = _DEFAULT_SETTINGS
         self.zeep_client_authless = ZeepAsyncClient(
-            wsdl=url,
+            wsdl=DocumentWithCache(url, self.transport, settings=settings),
             transport=self.transport,
             settings=settings,
             plugins=[WsAddressingPlugin()],
         )
         self.ws_client_authless = self.zeep_client_authless.create_service(
             binding_name, self.xaddr
         )
-
         self.zeep_client = ZeepAsyncClient(
-            wsdl=url,
+            wsdl=DocumentWithCache(url, self.transport, settings=settings),
             wsse=wsse,
             transport=self.transport,
             settings=settings,