NEW: implement ZeepBackend.with_params() ✨

combadge/support/zeep/backends/async_.py

@@ -1,13 +1,20 @@
 from __future__ import annotations
 
+from collections.abc import Collection
 from contextlib import AbstractAsyncContextManager
+from os import PathLike, fspath
+from ssl import SSLContext
 from types import TracebackType
 from typing import Any, Callable
 
+import httpx
 from pydantic import BaseModel
 from typing_extensions import Self
+from zeep import AsyncClient, Plugin
 from zeep.exceptions import Fault
 from zeep.proxy import AsyncOperationProxy, AsyncServiceProxy
+from zeep.transports import AsyncTransport
+from zeep.wsse import UsernameToken
 
 from combadge.core.backend import ServiceContainer
 from combadge.core.binder import BaseBoundService
@@ -19,7 +26,7 @@
 from combadge.support.shared.contextlib import asyncnullcontext
 from combadge.support.soap.request import Request
 from combadge.support.soap.response import SoapFaultT
-from combadge.support.zeep.backends.base import BaseZeepBackend
+from combadge.support.zeep.backends.base import BaseZeepBackend, ByBindingName, ByServiceName
 
 
 class ZeepBackend(
@@ -31,6 +38,58 @@ class ZeepBackend(
 
     __slots__ = ("_service", "_request_with", "_service_cache")
 
+    @classmethod
+    def with_params(
+        cls,
+        wsdl_path: PathLike,
+        *,
+        service: ByBindingName | ByServiceName | None = None,
+        plugins: Collection[Plugin] | None = None,
+        load_timeout: float | None = None,
+        operation_timeout: float | None = None,
+        wsse: UsernameToken | None = None,
+        verify_ssl: PathLike | bool | SSLContext = True,
+        cert: PathLike | tuple[PathLike, PathLike | None] | tuple[PathLike, PathLike | None, str | None] | None = None,
+    ) -> ZeepBackend:
+        """
+        Instantiate the backend using a set of the most common parameters.
+
+        Using the `__init__()` may become quite wordy, so this method simplifies typical use cases.
+        """
+        verify = verify_ssl if isinstance(verify_ssl, (bool, SSLContext)) else fspath(verify_ssl)
+
+        if isinstance(cert, tuple):
+            cert_file = cert[0]
+            key_file = cert[1]
+            password = cert[2] if len(cert) == 3 else None  # type: ignore[misc]
+            cert_ = (fspath(cert_file), fspath(key_file) if key_file else None, password)
+        elif cert is not None:
+            cert_ = fspath(cert)
+        else:
+            cert_ = None
+
+        transport = AsyncTransport(
+            timeout=None,  # overloaded
+            client=httpx.AsyncClient(timeout=operation_timeout, verify=verify, cert=cert_),
+            wsdl_client=httpx.Client(timeout=load_timeout, verify=verify, cert=cert_),
+        )
+
+        client = AsyncClient(fspath(wsdl_path), wsse=wsse, plugins=plugins, transport=transport)
+        if service is None:
+            service_proxy = client.service
+        elif isinstance(service, ByServiceName):
+            service_proxy = client.bind(service.service_name, service.port_name)
+        elif isinstance(service, ByBindingName):
+            # `create_service()` creates a sync service proxy, work around:
+            service_proxy = AsyncServiceProxy(
+                client,
+                client.wsdl.bindings[service.binding_name],
+                address=service.address,
+            )
+        else:
+            raise TypeError(type(service))
+        return cls(service_proxy)
+
     def __init__(
         self,
         service: AsyncServiceProxy,

combadge/support/zeep/backends/base.py

@@ -1,5 +1,8 @@
+from __future__ import annotations
+
 from abc import ABC
-from typing import Any, Generic, Tuple, Type, TypeVar, Union
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar, Union
 
 from pydantic import BaseModel, parse_obj_as
 from typing_extensions import get_args as get_type_args
@@ -33,7 +36,7 @@ def __init__(self, service: _ServiceProxyT) -> None:
         self._service = service
 
     @staticmethod
-    def _split_response_type(response_type: Type[Any]) -> Tuple[Type[BaseModel], Type[BaseSoapFault]]:
+    def _split_response_type(response_type: type[Any]) -> tuple[type[BaseModel], type[BaseSoapFault]]:
         """
         Split the response type into non-faults and faults.
 
@@ -70,11 +73,27 @@ def _get_operation(self, name: str) -> _OperationProxyT:
             raise RuntimeError(f"available operations are: {dir(self._service)}") from e
 
     @staticmethod
-    def _parse_response(value: CompoundValue, response_type: Type[ResponseT]) -> ResponseT:
+    def _parse_response(value: CompoundValue, response_type: type[ResponseT]) -> ResponseT:
         """Parse the response value using the generic response types."""
         return parse_obj_as(response_type, serialize_object(value, dict))
 
     @staticmethod
-    def _parse_soap_fault(exception: Fault, fault_type: Type[SoapFaultT]) -> SoapFaultT:
+    def _parse_soap_fault(exception: Fault, fault_type: type[SoapFaultT]) -> SoapFaultT:
         """Parse the SOAP fault."""
         return parse_obj_as(fault_type, exception.__dict__)
+
+
+@dataclass
+class ByBindingName:
+    """Create service by binding name and address."""
+
+    binding_name: str
+    address: str
+
+
+@dataclass
+class ByServiceName:
+    """Create service by service and port names."""
+
+    service_name: str | None = None
+    port_name: str | None = None

combadge/support/zeep/backends/sync.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 from contextlib import AbstractContextManager, nullcontext
+from os import PathLike, fspath
 from types import TracebackType
-from typing import Any, Callable
+from typing import Any, Callable, Collection
 
 from pydantic import BaseModel
 from typing_extensions import Self
+from zeep import Client, Plugin, Transport
 from zeep.exceptions import Fault
 from zeep.proxy import OperationProxy, ServiceProxy
+from zeep.wsse import UsernameToken
 
 from combadge.core.backend import ServiceContainer
 from combadge.core.binder import BaseBoundService
@@ -18,14 +21,54 @@
 from combadge.support.shared.sync import SupportsRequestWith
 from combadge.support.soap.request import Request
 from combadge.support.soap.response import SoapFaultT
-from combadge.support.zeep.backends.base import BaseZeepBackend
+from combadge.support.zeep.backends.base import BaseZeepBackend, ByBindingName, ByServiceName
 
 
 class ZeepBackend(BaseZeepBackend[ServiceProxy, OperationProxy], SupportsRequestWith[Request], ServiceContainer):
     """Synchronous Zeep service."""
 
     __slots__ = ("_service", "_request_with", "_service_cache")
 
+    @classmethod
+    def with_params(
+        cls,
+        wsdl_path: PathLike,
+        *,
+        service: ByBindingName | ByServiceName | None = None,
+        plugins: Collection[Plugin] | None = None,
+        load_timeout: float | None = None,
+        operation_timeout: float | None = None,
+        wsse: UsernameToken | None = None,
+        verify_ssl: bool | PathLike = True,
+        cert_file: PathLike | None = None,
+        key_file: PathLike | None = None,
+    ) -> ZeepBackend:
+        """
+        Instantiate the backend using a set of the most common parameters.
+
+        Using the `__init__()` may become quite wordy, so this method simplifies typical use cases.
+        """
+        client = Client(
+            fspath(wsdl_path),
+            wsse=wsse,
+            transport=Transport(timeout=load_timeout, operation_timeout=operation_timeout),
+            plugins=plugins,
+        )
+        client.transport.session.verify = verify_ssl if isinstance(verify_ssl, bool) else fspath(verify_ssl)
+        client.transport.session.cert = (
+            fspath(cert_file) if cert_file is not None else None,
+            fspath(key_file) if key_file is not None else None,
+        )
+        if service is None:
+            service_proxy = client.service
+        elif isinstance(service, ByServiceName):
+            service_proxy = client.bind(service.service_name, service.port_name)
+        elif isinstance(service, ByBindingName):
+            service_proxy = client.create_service(service.binding_name, service.address)
+        else:
+            raise TypeError(type(service))
+        return cls(service_proxy)
+
     def __init__(
         self,
         service: ServiceProxy,

tests/integration/cassettes/test_number_conversion/test_happy_path_with_params.yaml

@@ -0,0 +1,55 @@
+interactions:
+- request:
+    body: '<?xml version=''1.0'' encoding=''utf-8''?>
+
+      <soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/"><soap-env:Body><ns0:NumberToWords
+      xmlns:ns0="http://www.dataaccess.com/webservicesserver/"><ns0:ubiNum>42</ns0:ubiNum></ns0:NumberToWords></soap-env:Body></soap-env:Envelope>'
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '291'
+      Content-Type:
+      - text/xml; charset=utf-8
+      SOAPAction:
+      - '""'
+      User-Agent:
+      - Zeep/4.2.1 (www.python-zeep.org)
+    method: POST
+    uri: https://www.dataaccess.com/webservicesserver/NumberConversion.wso
+  response:
+    body:
+      string: !!binary |
+        H4sIAAAAAAAEAO29B2AcSZYlJi9tynt/SvVK1+B0oQiAYBMk2JBAEOzBiM3mkuwdaUcjKasqgcpl
+        VmVdZhZAzO2dvPfee++999577733ujudTif33/8/XGZkAWz2zkrayZ4hgKrIHz9+fB8/Ih7/Hu8W
+        ZXqZ101RLT/7aHe881GaL6fVrFhefPbRuj3fPvjo9zj6jZPHTZWtHp0uL/OyWuUpvbRsHuGzzz6a
+        t+3q0d27zXSeL7JmTF/h83FVX9zFL3dzfenuRwQnTQXSk2p2zX/SB4tHL9aLSV6/qb5b1bPmVd6s
+        qmVjOlnYHq6ursazrM2y6TRvmvG0Wty9yidNXl8W+IB+5rV2gicCd122R+dV3V6n7VWVPr4bbyEA
+        Yt8yXvz9Yx6bGYb+Zchz9P8A3rWog1YBAAA=
+    headers:
+      Cache-Control:
+      - private, max-age=0
+      Content-Encoding:
+      - gzip
+      Content-Length:
+      - '311'
+      Content-Type:
+      - text/xml; charset=utf-8
+      Date:
+      - Fri, 28 Jul 2023 12:21:43 GMT
+      Server:
+      - Server
+      Vary:
+      - Accept-Encoding
+      Web-Service:
+      - DataFlex 19.1
+      X-Powered-By:
+      - ASP.NET
+    status:
+      code: 200
+      message: OK
+version: 1

tests/integration/cassettes/test_number_conversion/test_happy_path_with_params_async.yaml

@@ -0,0 +1,52 @@
+interactions:
+- request:
+    body: '<?xml version=''1.0'' encoding=''utf-8''?>
+
+      <soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/"><soap-env:Body><ns0:NumberToWords
+      xmlns:ns0="http://www.dataaccess.com/webservicesserver/"><ns0:ubiNum>42</ns0:ubiNum></ns0:NumberToWords></soap-env:Body></soap-env:Envelope>'
+    headers:
+      accept:
+      - '*/*'
+      accept-encoding:
+      - gzip, deflate
+      connection:
+      - keep-alive
+      content-length:
+      - '291'
+      content-type:
+      - text/xml; charset=utf-8
+      host:
+      - www.dataaccess.com
+      soapaction:
+      - '""'
+      user-agent:
+      - Zeep/4.2.1 (www.python-zeep.org)
+    method: POST
+    uri: https://www.dataaccess.com/webservicesserver/NumberConversion.wso
+  response:
+    content: "<?xml version=\"1.0\" encoding=\"utf-8\"?>\r\n<soap:Envelope xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\">\r\n
+      \ <soap:Body>\r\n    <m:NumberToWordsResponse xmlns:m=\"http://www.dataaccess.com/webservicesserver/\">\r\n
+      \     <m:NumberToWordsResult>forty two </m:NumberToWordsResult>\r\n    </m:NumberToWordsResponse>\r\n
+      \ </soap:Body>\r\n</soap:Envelope>"
+    headers:
+      Cache-Control:
+      - private, max-age=0
+      Content-Encoding:
+      - gzip
+      Content-Length:
+      - '311'
+      Content-Type:
+      - text/xml; charset=utf-8
+      Date:
+      - Fri, 28 Jul 2023 12:22:51 GMT
+      Server:
+      - Server
+      Vary:
+      - Accept-Encoding
+      Web-Service:
+      - DataFlex 19.1
+      X-Powered-By:
+      - ASP.NET
+    http_version: HTTP/1.1
+    status_code: 200
+version: 1

tests/integration/test_number_conversion.py

@@ -11,6 +11,7 @@
 from combadge.core.response import ErrorResponse, SuccessfulResponse
 from combadge.support.soap.markers import Body, operation_name
 from combadge.support.zeep.backends.async_ import ZeepBackend as AsyncZeepBackend
+from combadge.support.zeep.backends.base import ByServiceName
 from combadge.support.zeep.backends.sync import ZeepBackend as SyncZeepBackend
 
 
@@ -88,3 +89,27 @@ async def test_happy_path_scalar_response_async(number_conversion_service_async:
     assert_type(response, NumberToWordsResponse)
 
     assert response.__root__ == "forty two "
+
+
+@mark.vcr
+def test_happy_path_with_params() -> None:
+    backend = SyncZeepBackend.with_params(
+        Path(__file__).parent / "wsdl" / "NumberConversion.wsdl",
+        service=ByServiceName(port_name="NumberConversionSoap"),
+        operation_timeout=1,
+    )
+    service = SupportsNumberConversion.bind(backend)
+    response = service.number_to_words(NumberToWordsRequest(number=42)).unwrap()
+    assert response.__root__ == "forty two "
+
+
+@mark.vcr
+async def test_happy_path_with_params_async() -> None:
+    backend = AsyncZeepBackend.with_params(
+        Path(__file__).parent / "wsdl" / "NumberConversion.wsdl",
+        service=ByServiceName(port_name="NumberConversionSoap"),
+        operation_timeout=1,
+    )
+    service = SupportsNumberConversionAsync.bind(backend)
+    response = (await service.number_to_words(NumberToWordsRequest(number=42))).unwrap()
+    assert response.__root__ == "forty two "