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 
 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
112
class ZeepBackend(BaseZeepBackend[ServiceProxy, OperationProxy], ServiceContainer):
    """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)
        ServiceContainer.__init__(self)

    def bind_method(self, signature: Signature) -> ServiceMethod[ZeepBackend]:  # noqa: D102
        response_type, fault_type = self._adapt_response_type(signature.return_type)
        backend = self

        def bound_method(self: BaseBoundService[ZeepBackend], *args: Any, **kwargs: Any) -> Any:
            request = signature.build_request(Request, self, args, kwargs)
            operation = backend._get_operation(request.get_operation_name())
            try:
                response = operation(**(request.payload or {}))
            except Fault as e:
                return 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 
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
@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 
69
70
71
72
73
74
75
76
77
78
79
80
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)
    ServiceContainer.__init__(self)

Async

Asynchronous Zeep service.

Source code in combadge/support/zeep/backends/async_.py 
 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
127
128
129
130
class ZeepBackend(
    BaseZeepBackend[AsyncServiceProxy, AsyncOperationProxy],
    ServiceContainer,
):
    """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)
        ServiceContainer.__init__(self)

    def bind_method(self, signature: Signature) -> ServiceMethod[ZeepBackend]:  # noqa: D102
        response_type, fault_type = self._adapt_response_type(signature.return_type)
        backend = self

        async def bound_method(self: BaseBoundService[ZeepBackend], *args: Any, **kwargs: Any) -> Any:
            request = signature.build_request(Request, self, args, kwargs)
            operation = backend._get_operation(request.get_operation_name())
            try:
                response = await operation(**(request.payload or {}))
            except Fault as e:
                return 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 
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
@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 
87
88
89
90
91
92
93
94
95
96
97
98
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)
    ServiceContainer.__init__(self)

Binding specification

ByBindingName dataclass

Create service by binding name and address.

Examples:

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

)

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

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

    binding_name: str
    address: HttpUrl | str

    @property
    def address_string(self) -> str:
        """Return the service address as a plain `#!python str`."""
        if isinstance(self.address, 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 
131
132
133
134
135
136
137
138
139
140
141
@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