Skip to content

Zeep

Support for Zeep client – requires combadge[zeep] extra.

Sync

Synchronous Zeep service.

Source code in combadge/support/zeep/backends/sync.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class ZeepBackend(BaseZeepBackend[ServiceProxy, OperationProxy]):
    """Synchronous Zeep service."""

    __slots__ = ("_service", "_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_string)
        else:
            raise TypeError(type(service))
        return cls(service_proxy)

    def __init__(
        self,
        service: ServiceProxy,
    ) -> None:
        """
        Instantiate the backend.

        Args:
            service: [service proxy object](https://docs.python-zeep.org/en/master/client.html#the-serviceproxy-object)
        """
        BaseZeepBackend.__init__(self, service)

    @classmethod
    @override
    def bind_method(cls, signature: Signature, /) -> ServiceMethod[ZeepBackend]:  # noqa: D102
        response_type, fault_type = cls._adapt_response_type(signature.return_type)

        def bound_method(self: BaseBoundService[ZeepBackend], *args: Any, **kwargs: Any) -> Any:
            request = signature.build_request(Request, self, args, kwargs)
            operation = self.__combadge_backend__._get_operation(request.get_operation_name())
            try:
                response = operation(**(request.payload or {}), _soapheaders=request.soap_header)
            except Fault as e:
                return self.__combadge_backend__._parse_soap_fault(e, fault_type)
            except Exception as e:
                raise BackendError(e) from e
            else:
                return signature.apply_response_markers(response, serialize_object(response, dict), response_type)

        return bound_method  # type: ignore[return-value]

    binder = bind_method  # type: ignore[assignment]

    def __enter__(self) -> Self:
        self._service = self._service.__enter__()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> Any:
        return self._service.__exit__(exc_type, exc_value, traceback)

with_params classmethod 

with_params(
    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.

Source code in combadge/support/zeep/backends/sync.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@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_string)
    else:
        raise TypeError(type(service))
    return cls(service_proxy)

__init__ 

__init__(service: ServiceProxy) -> None

Instantiate the backend.

Parameters:

Name Type Description Default
service ServiceProxy

service proxy object

 required
Source code in combadge/support/zeep/backends/sync.py 
68
69
70
71
72
73
74
75
76
77
78
def __init__(
    self,
    service: ServiceProxy,
) -> None:
    """
    Instantiate the backend.

    Args:
        service: [service proxy object](https://docs.python-zeep.org/en/master/client.html#the-serviceproxy-object)
    """
    BaseZeepBackend.__init__(self, service)

Async

Asynchronous Zeep service.

Source code in combadge/support/zeep/backends/async_.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class ZeepBackend(BaseZeepBackend[AsyncServiceProxy, AsyncOperationProxy]):
    """Asynchronous Zeep service."""

    __slots__ = ("_service", "_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_string,
            )
        else:
            raise TypeError(type(service))
        return cls(service_proxy)

    def __init__(
        self,
        service: AsyncServiceProxy,
    ) -> None:
        """
        Instantiate the backend.

        Args:
            service: [service proxy object](https://docs.python-zeep.org/en/master/client.html#the-serviceproxy-object)
        """
        BaseZeepBackend.__init__(self, service)

    @classmethod
    @override
    def bind_method(cls, signature: Signature, /) -> ServiceMethod[ZeepBackend]:  # noqa: D102
        response_type, fault_type = cls._adapt_response_type(signature.return_type)

        async def bound_method(self: BaseBoundService[ZeepBackend], *args: Any, **kwargs: Any) -> Any:
            request = signature.build_request(Request, self, args, kwargs)
            operation = self.__combadge_backend__._get_operation(request.get_operation_name())
            try:
                response = await operation(**(request.payload or {}), _soapheaders=request.soap_header)
            except Fault as e:
                return self.__combadge_backend__._parse_soap_fault(e, fault_type)
            except Exception as e:
                raise BackendError(e) from e
            else:
                return signature.apply_response_markers(response, serialize_object(response, dict), response_type)

        return bound_method  # type: ignore[return-value]

    binder = bind_method  # type: ignore[assignment]

    async def __aenter__(self) -> Self:
        self._service = await self._service.__aenter__()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> Any:
        return await self._service.__aexit__(exc_type, exc_value, traceback)

with_params classmethod 

with_params(
    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.

Source code in combadge/support/zeep/backends/async_.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@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_string,
        )
    else:
        raise TypeError(type(service))
    return cls(service_proxy)

__init__ 

__init__(service: AsyncServiceProxy) -> None

Instantiate the backend.

Parameters:

Name Type Description Default
service AsyncServiceProxy

service proxy object

 required
Source code in combadge/support/zeep/backends/async_.py 
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self,
    service: AsyncServiceProxy,
) -> None:
    """
    Instantiate the backend.

    Args:
        service: [service proxy object](https://docs.python-zeep.org/en/master/client.html#the-serviceproxy-object)
    """
    BaseZeepBackend.__init__(self, service)

Binding specification

ByBindingName dataclass

Create service by binding name and address.

Examples:

>>> ByBindingName(
>>>     binding_name="{http://www.dataaccess.com/webservicesserver/}NumberConversionSoapBinding",
>>>     address=HttpUrl("https://www.dataaccess.com/webservicesserver/NumberConversion.wso",
>>> )

)

Source code in combadge/support/zeep/backends/base.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@dataclass(**SLOTS)
class ByBindingName:
    """
    Create service by binding name and address.

    Examples:
        >>> ByBindingName(
        >>>     binding_name="{http://www.dataaccess.com/webservicesserver/}NumberConversionSoapBinding",
        >>>     address=HttpUrl("https://www.dataaccess.com/webservicesserver/NumberConversion.wso",
        >>> )
    )
    """

    binding_name: str
    address: HttpUrl | Url | str

    @property
    def address_string(self) -> str:
        """Return the service address as a plain `#!python str`."""
        if isinstance(self.address, (HttpUrl, Url)):
            return str(self.address)
        return self.address

address_string property 

address_string: str

Return the service address as a plain str.

ByServiceName dataclass

Create service by service and port names.

Examples:

>>> ByServiceName(port_name="NumberConversionSoap")
Source code in combadge/support/zeep/backends/base.py 
135
136
137
138
139
140
141
142
143
144
145
@dataclass(**SLOTS)
class ByServiceName:
    """
    Create service by service and port names.

    Examples:
        >>> ByServiceName(port_name="NumberConversionSoap")
    """

    service_name: str | None = None
    port_name: str | None = None