ds_protocol_soap_py_lib.linked_service¶

File: __init__.py Region: ds_protocol_soap_py_lib/linked_service

SOAP Linked Service

This module implements a linked service for SOAP APIs.

Example

>>> import uuid
>>> from ds_protocol_soap_py_lib.enums import AuthType
>>> from ds_protocol_soap_py_lib.linked_service import (
...     BasicAuthSettings,
...     SoapLinkedService,
...     SoapLinkedServiceSettings,
... )
>>> linked_service = SoapLinkedService(
...     id=uuid.uuid4(),
...     name="example::linked_service",
...     version="1.0.0",
...     settings=SoapLinkedServiceSettings(
...         wsdl="https://api.example.com?WSDL",
...         auth_type=AuthType.BASIC,
...         basic=BasicAuthSettings(
...             username="user",
...             password="pass",
...         ),
...     ),
... )
>>> linked_service.connect()

Submodules¶

ds_protocol_soap_py_lib.linked_service.soap

Classes¶

`BasicAuthSettings`	Settings for SOAP Basic authentication.
`ParameterBasedAuthSettings`	Settings for parameter-based authentication.
`SoapLinkedService`	Linked service for SOAP APIs.
`SoapLinkedServiceSettings`	Settings for SOAP linked service connections.

Package Contents¶

class ds_protocol_soap_py_lib.linked_service.BasicAuthSettings[source]¶

Settings for SOAP Basic authentication.

Uses standard HTTP Basic auth with base64-encoded username:password on the underlying requests transport.

username: str¶: The username for basic auth.

password: str¶: The password for basic auth.

class ds_protocol_soap_py_lib.linked_service.ParameterBasedAuthSettings[source]¶

Settings for parameter-based authentication.

Uses custom parameters passed in the SOAP body rather than HTTP headers. Access the resolved parameters via SoapLinkedService.body_auth_params.

auth_param_key1: str¶: The key for the first parameter used for authentication in the SOAP body.

auth_param_value1: str¶: The value for the first parameter used for authentication in the SOAP body.

auth_param_key2: str | None = None¶: Additional auth parameter key (optional).

auth_param_value2: str | None = None¶: Additional auth parameter value (optional).

auth_param_key3: str | None = None¶: Additional auth parameter key (optional).

auth_param_value3: str | None = None¶: Additional auth parameter value (optional).

class ds_protocol_soap_py_lib.linked_service.SoapLinkedService[source]¶

Bases: ds_resource_plugin_py_lib.common.resource.linked_service.LinkedService[SoapLinkedServiceSettingsType], Generic[SoapLinkedServiceSettingsType]

Linked service for SOAP APIs.

Wraps a zeep.Client and handles WSDL loading, transport setup, and authentication. The client is available via the connection property after connect() is called.

Supports use as a context manager:

with linked_service:
    result = linked_service.connection.service.SomeMethod(...)

settings: SoapLinkedServiceSettingsType¶

_client: zeep.Client | None = None¶

_credential: str | None = None¶

property type: ds_protocol_soap_py_lib.enums.ResourceType¶

Get the type of the linked service.

Returns:: ResourceType

property connection: zeep.Client¶

Return the zeep Client established by connect().

Returns:: zeep.Client
Raises:: ConnectionError – If connect() has not been called.

property body_auth_params: dict[str, str]¶

Return keyword authentication parameters for SOAP method calls.

For BASIC_WITH_TOKEN_EXCHANGE auth, returns {credential_param_key: credential}. For PARAMETER_BASED auth, returns the configured body parameters.

Returns:: dict[str, str]
Raises:: ConnectionError – If auth_type is BASIC_WITH_TOKEN_EXCHANGE and no credential is set (i.e. connect() has not been called).

_init_client() → zeep.Client[source]¶

Initialise the zeep Client from the WSDL and configured transport/settings.

Returns:: zeep.Client
Raises:: ConnectionError – If the WSDL cannot be reached or parsed.

_configure_basic_auth(client: zeep.Client) → None[source]¶

Configure HTTP Basic authentication on the transport session.

Parameters:: client – The zeep Client to configure.
Raises:: LinkedServiceException – If basic auth settings are missing.

_configure_basic_with_token_exchange_auth(client: zeep.Client) → None[source]¶

Exchange credentials for a session token via a dedicated auth WSDL.

Opens a temporary HTTP Basic Auth session against auth_wsdl, calls auth_method to retrieve a session token, stores it in _credential, then closes the auth session. The token is injected into all subsequent data calls via body_auth_params. The client argument is unused — the data WSDL client is not involved in the exchange.

Parameters:: client – Unused. Present for dispatch-table consistency with other auth handlers.
Raises:: LinkedServiceException – If basic_with_token_exchange settings are missing or the credential exchange call fails.

_configure_parameter_based_auth(client: zeep.Client) → None[source]¶

Validate that parameter-based auth settings are present.

Parameters are not applied to the transport — they are passed per-call via body_auth_params.

Parameters:: client – The zeep Client (unused; present for dispatch-table consistency).
Raises:: LinkedServiceException – If parameter-based auth settings are missing.

connect() → None[source]¶

Load the WSDL and configure authentication.

Stores the initialised zeep.Client internally so it is accessible via the connection property. Verifies the connection by calling auth_test_method before returning. Safe to call multiple times — re-connects on each call.

Returns:

None

Raises:

ConnectionError – If the WSDL cannot be reached or parsed.
LinkedServiceException – If auth settings are missing, auth_type is unsupported, or the connection test call fails (wrong credentials, method, or parameters).

test_connection() → tuple[bool, str][source]¶

Verify the connection to the SOAP API by calling auth_test_method.

Does not raise on failure — returns (False, reason) instead. Returns (False, "No auth_test_method configured") if auth_test_method is not set.

Returns:: (True, "") on success, (False, reason) on failure.
Return type:: tuple[bool, str]

close() → None[source]¶

Close the underlying requests session, release the zeep Client, and clear the session credential.

Safe to call multiple times.

_build_body_auth_params() → dict[str, str][source]¶

Build the authentication parameter dict for parameter-based auth.

Returns:: dict[str, str]

class ds_protocol_soap_py_lib.linked_service.SoapLinkedServiceSettings[source]¶

Bases: ds_resource_plugin_py_lib.common.resource.linked_service.LinkedServiceSettings

Settings for SOAP linked service connections.

Provide the appropriate auth settings object based on your auth_type:

AuthType.BASIC → basic
AuthType.BASIC_WITH_TOKEN_EXCHANGE → basic_with_token_exchange
AuthType.PARAMETER_BASED → parameter_based

Example

>>> settings = SoapLinkedServiceSettings(
...     wsdl="https://api.example.com?WSDL",
...     auth_type=AuthType.BASIC,
...     auth_test_method="SomeHealthCheckMethod",
...     basic=BasicAuthSettings(
...         username="user",
...         password="pass",
...     ),
... )

wsdl: str¶: The WSDL endpoint URL.

auth_type: ds_protocol_soap_py_lib.enums.AuthType¶: The authentication type to use.

auth_test_method: str | None = None¶: The SOAP operation name used to verify the connection during connect() and test_connection(). Authentication in SOAP happens at call time, so a real method must be invoked to verify credentials. When None, the connection test is skipped and credentials are not verified until the first real call.

auth_test_method_params: dict[str, Any]¶: Optional parameters to include when calling the auth_test_method during connection testing. Useful if the method requires additional non-auth parameters.

basic: BasicAuthSettings | None = None¶: Settings for Basic authentication. Required when auth_type=AuthType.BASIC.

basic_with_token_exchange: BasicWithTokenExchangeAuthSettings | None = None¶: Settings for Basic + token exchange authentication. Required when auth_type=AuthType.BASIC_WITH_TOKEN_EXCHANGE.

parameter_based: ParameterBasedAuthSettings | None = None¶: Settings for parameter-based authentication. Required when auth_type=AuthType.PARAMETER_BASED.

cache: zeep.cache.Base | None = None¶: Optional zeep cache backend. Defaults to no cache.

timeout: int | float = 300¶: Timeout in seconds for WSDL loading and SOAP calls. Defaults to 300.

operation_timeout: int | float | None = None¶: Timeout in seconds for individual SOAP operations. Defaults to timeout.

strict: bool = True¶: Raise errors on WSDL non-conformance. Defaults to True.

raw_response: bool = False¶: Return the raw requests response instead of parsed objects. Defaults to False.

forbid_dtd: bool = False¶: Forbid DTD in XML responses. Defaults to False.

forbid_entities: bool = True¶: Forbid external entity references in XML. Defaults to True.

forbid_external: bool = True¶: Forbid external resource access in XML. Defaults to True.

xml_huge_tree: bool = False¶: Enable lxml huge_tree option for very large XML responses. Defaults to False.

force_https: bool = True¶: Require HTTPS for SOAP calls. Defaults to True.

extra_http_headers: dict[str, str] | None = None¶: Additional HTTP headers to include in every request. Defaults to None.

xsd_ignore_sequence_order: bool = False¶: Ignore XSD sequence ordering constraints. Defaults to False.