class worldline.connect.sdk.api_resource.ApiResource(parent: ApiResource | None = None, communicator: Communicator | None = None, path_context: Mapping[str, str] | None = None, client_meta_info: str | None = None)

Bases: object

Base class of all Worldline Global Collect platform API resources.

__init__(parent: ApiResource | None = None, communicator: Communicator | None = None, path_context: Mapping[str, str] | None = None, client_meta_info: str | None = None)

The parent and/or communicator must be given.

class worldline.connect.sdk.call_context.CallContext(idempotence_key: str | None = None)

Bases: object

A call context can be used to send extra information with a request, and to receive extra information from a response.

Please note that this class is not thread-safe. Each request should get its own call context instance.

__annotations__ = {'_CallContext__idempotence_key': typing.Optional[str], '_CallContext__idempotence_request_timestamp': typing.Optional[int]}

__init__(idempotence_key: str | None = None)

Sets the idempotence key to use for the next request for which this call context is used.

property idempotence_key: str | None

Returns:

The idempotence key.

property idempotence_request_timestamp: int | None

Returns:

The idempotence request timestamp from the response to the last request for which this call context was used, or None if no idempotence request timestamp was present.

class worldline.connect.sdk.client.Client(communicator: Communicator, client_meta_info: str | None = None)

Bases: ApiResource, LoggingCapable, ObfuscationCapable

Worldline Global Collect platform client.

This client and all its child clients are bound to one specific value for the X-GCS-ClientMetaInfo header. To get a new client with a different header value, use with_client_meta_info.

Thread-safe.

__abstractmethods__ = frozenset({})

__annotations__ = {}

__enter__()

__exit__(exc_type, exc_val, exc_tb)

__init__(communicator: Communicator, client_meta_info: str | None = None)

Parameters:

• communicator – worldline.connect.sdk.communicator.Communicator

• client_meta_info – str

close() → None

Releases any system resources associated with this object.

close_expired_connections() → None

Utility method that delegates the call to this client’s communicator.

close_idle_connections(idle_time: timedelta) → None

Utility method that delegates the call to this client’s communicator.

Parameters:

idle_time – a datetime.timedelta object indicating the idle time

disable_logging() → None

Turns off logging.

enable_logging(communicator_logger: CommunicatorLogger) → None

Turns on logging using the given communicator logger.

Raises:

ValueError – If the given communicator logger is None.

set_body_obfuscator(body_obfuscator: BodyObfuscator) → None

Sets the current body obfuscator to use.

set_header_obfuscator(header_obfuscator: HeaderObfuscator) → None

Sets the current header obfuscator to use.

v1() → V1Client

with_client_meta_info(client_meta_info: str | None) → Client

Parameters:

client_meta_info – JSON string containing the metadata for the client

Returns:

a new Client which uses the passed metadata for the X-GCS-ClientMetaInfo header.

Raises:

MarshallerSyntaxException – if the given clientMetaInfo is not a valid JSON string

class worldline.connect.sdk.communicator.Communicator(api_endpoint: str | ParseResult, connection: Connection, authenticator: Authenticator, metadata_provider: MetadataProvider, marshaller: Marshaller)

Bases: LoggingCapable, ObfuscationCapable

Used to communicate with the Worldline Global Collect platform web services.

It contains all the logic to transform a request object to an HTTP request and an HTTP response to a response object.

__abstractmethods__ = frozenset({})

__annotations__ = {}

__enter__()

__exit__(exc_type, exc_val, exc_tb)

close() → None

Releases any system resources associated with this object.

close_expired_connections() → None

Utility method that delegates the call to this communicator’s connection if that’s an instance of PooledConnection. If not this method does nothing.

close_idle_connections(idle_time: timedelta) → None

Utility method that delegates the call to this communicator’s connection if that’s an instance of PooledConnection. If not this method does nothing.

Parameters:

idle_time – a datetime.timedelta object indicating the idle time

delete(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, response_type: Type[T], context: CallContext | None) → T

Corresponds to the HTTP DELETE method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• response_type – The type of response to return.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

delete_with_binary_response(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, context: CallContext | None) → Tuple[Mapping[str, str], Iterable[bytes]]

Corresponds to the HTTP DELETE method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

disable_logging() → None

Turns off logging.

enable_logging(communicator_logger: CommunicatorLogger) → None

Turns on logging using the given communicator logger.

Raises:

ValueError – If the given communicator logger is None.

get(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, response_type: Type[T], context: CallContext | None) → T

Corresponds to the HTTP GET method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• response_type – The type of response to return.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

get_with_binary_response(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, context: CallContext | None) → Tuple[Mapping[str, str], Iterable[bytes]]

Corresponds to the HTTP GET method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

property marshaller: Marshaller

Returns:

The Marshaller object associated with this communicator. Never None.

post(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, request_body: Any, response_type: Type[T], context: CallContext | None) → T

Corresponds to the HTTP POST method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• request_body – The optional request body to send.

• response_type – The type of response to return.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

post_with_binary_response(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, request_body: Any, context: CallContext | None) → Tuple[Mapping[str, str], Iterable[bytes]]

Corresponds to the HTTP POST method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• request_body – The optional request body to send.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

put(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, request_body: Any, response_type: Type[T], context: CallContext | None) → T

Corresponds to the HTTP PUT method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• request_body – The optional request body to send.

• response_type – The type of response to return.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

put_with_binary_response(relative_path: str, request_headers: List[RequestHeader] | None, request_parameters: ParamRequest | None, request_body: Any, context: CallContext | None) → Tuple[Mapping[str, str], Iterable[bytes]]

Corresponds to the HTTP PUT method.

Parameters:

• relative_path – The path to call, relative to the base URI.

• request_headers – An optional list of request headers.

• request_parameters – An optional set of request parameters.

• request_body – The optional request body to send.

• context – The optional call context to use.

Raises:

• CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

• ResponseException – when an error response was received from the Worldline Global Collect platform

set_body_obfuscator(body_obfuscator: BodyObfuscator) → None

Sets the current body obfuscator to use.

set_header_obfuscator(header_obfuscator: HeaderObfuscator) → None

Sets the current header obfuscator to use.

class worldline.connect.sdk.communicator_configuration.CommunicatorConfiguration(properties: ConfigParser | None = None, api_endpoint: str | ParseResult | None = None, authorization_id: str | None = None, authorization_secret: str | None = None, api_key_id: str | None = None, secret_api_key: str | None = None, authorization_type: str | None = None, connect_timeout: int | None = None, socket_timeout: int | None = None, max_connections: int | None = None, proxy_configuration: ProxyConfiguration | None = None, integrator: str | None = None, shopping_cart_extension: ShoppingCartExtension | None = None)

Bases: object

Configuration for the communicator.

DEFAULT_MAX_CONNECTIONS = 10

__annotations__ = {'_CommunicatorConfiguration__api_endpoint': typing.Optional[urllib.parse.ParseResult], '_CommunicatorConfiguration__authorization_id': typing.Optional[str], '_CommunicatorConfiguration__authorization_secret': typing.Optional[str], '_CommunicatorConfiguration__authorization_type': typing.Optional[str], '_CommunicatorConfiguration__connect_timeout': typing.Optional[int], '_CommunicatorConfiguration__integrator': typing.Optional[str], '_CommunicatorConfiguration__max_connections': typing.Optional[int], '_CommunicatorConfiguration__proxy_configuration': typing.Optional[worldline.connect.sdk.proxy_configuration.ProxyConfiguration], '_CommunicatorConfiguration__shopping_cart_extension': typing.Optional[worldline.connect.sdk.domain.shopping_cart_extension.ShoppingCartExtension], '_CommunicatorConfiguration__socket_timeout': typing.Optional[int], '__api_endpoint': 'Optional[ParseResult]', '__authorization_id': 'Optional[str]', '__authorization_secret': 'Optional[str]', '__authorization_type': 'Optional[str]', '__connect_timeout': 'Optional[int]', '__integrator': 'Optional[str]', '__max_connections': 'Optional[int]', '__proxy_configuration': 'Optional[ProxyConfiguration]', '__shopping_cart_extension': 'Optional[ShoppingCartExtension]', '__socket_timeout': 'Optional[int]'}

__init__(properties: ConfigParser | None = None, api_endpoint: str | ParseResult | None = None, authorization_id: str | None = None, authorization_secret: str | None = None, api_key_id: str | None = None, secret_api_key: str | None = None, authorization_type: str | None = None, connect_timeout: int | None = None, socket_timeout: int | None = None, max_connections: int | None = None, proxy_configuration: ProxyConfiguration | None = None, integrator: str | None = None, shopping_cart_extension: ShoppingCartExtension | None = None)

Parameters:

• properties – a ConfigParser.ConfigParser object containing configuration data

• connect_timeout – connection timeout for the network communication in seconds

• socket_timeout – socket timeout for the network communication in seconds

• max_connections – The maximum number of connections in the connection pool

property api_endpoint: ParseResult | None

The Worldline Global Collect platform API endpoint URI.

property api_key_id: str | None

An identifier for the secret API key. The api_key_id can be retrieved from the Configuration Center. This identifier is visible in the HTTP request and is also used to identify the correct account.

This property is an alias for authorization_id

property authorization_id: str | None

An id used for authorization. The meaning of this id is different for each authorization type. For instance, for v1HMAC this is the identifier for the secret API key.

property authorization_secret: str | None

A secret used for authorization. The meaning of this secret is different for each authorization type. For instance, for v1HMAC this is the secret API key.

property authorization_type: str | None

property connect_timeout: int | None

Connection timeout for the underlying network communication in seconds

property integrator: str | None

property max_connections: int | None

property proxy_configuration: ProxyConfiguration | None

property secret_api_key: str | None

A shared secret. The shared secret can be retrieved from the Configuration Center. An api_key_id and secret_api_key always go hand-in-hand, the difference is that secret_api_key is never visible in the HTTP request. This secret is used as input for the HMAC algorithm.

This property is an alias for authorization_secret

property shopping_cart_extension: ShoppingCartExtension | None

property socket_timeout: int | None

Socket timeout for the underlying network communication in seconds

class worldline.connect.sdk.factory.Factory

Bases: object

Worldline Global Collect platform factory for several SDK components.

static create_client_from_communicator(communicator: Communicator) → Client

Create a Client based on the settings stored in the Communicator argument

static create_client_from_configuration(communicator_configuration: CommunicatorConfiguration, metadata_provider: MetadataProvider | None = None, connection: Connection | None = None, authenticator: Authenticator | None = None, marshaller: Marshaller | None = None) → Client

Create a Client based on the configuration stored in the CommunicatorConfiguration argument

static create_client_from_file(configuration_file_name: str | bytes, authorization_id: str, authorization_secret: str, metadata_provider: MetadataProvider | None = None, connection: Connection | None = None, authenticator: Authenticator | None = None, marshaller: Marshaller | None = None) → Client

Creates a Client based on the configuration values in configuration_file_name, authorization_id and authorization_secret.

static create_communicator_from_configuration(communicator_configuration: CommunicatorConfiguration, metadata_provider: MetadataProvider | None = None, connection: Connection | None = None, authenticator: Authenticator | None = None, marshaller: Marshaller | None = None) → Communicator

Creates a Communicator based on the configuration stored in the CommunicatorConfiguration argument

static create_communicator_from_file(configuration_file_name: str | bytes, authorization_id: str, authorization_secret: str, metadata_provider: MetadataProvider | None = None, connection: Connection | None = None, authenticator: Authenticator | None = None, marshaller: Marshaller | None = None) → Communicator

Creates a Communicator based on the configuration values in configuration_file_name, api_id_key and authorization_secret.

static create_configuration(configuration_file_name: str | bytes, authorization_id: str, authorization_secret: str) → CommunicatorConfiguration

Creates a CommunicatorConfiguration based on the configuration values in configuration_file_name, authorization_id and authorization_secret.

class worldline.connect.sdk.proxy_configuration.ProxyConfiguration(host: str, port: int, scheme: str = 'http', username: str | None = None, password: str | None = None)

Bases: object

HTTP proxy configuration.

Can be initialised directly from a host and port or can be constructed from a uri using fromuri

__str__()

Return a proxy string in the form scheme://username:password@host:port or scheme://host:port if authentication is absent

Supports HTTP Basic Auth

static from_uri(uri: str, username: str | None = None, password: str | None = None) → ProxyConfiguration

Constructs a ProxyConfiguration from a URI; if username and/or password are given they will be used instead of corresponding data in the URI

property host: str

property password: str | None

property port: int

property scheme: str

property username: str | None

class worldline.connect.sdk.authentication.authenticator.Authenticator

Bases: ABC

Used to authenticate requests to the Worldline Global Collect platform.

__abstractmethods__ = frozenset({'get_authorization'})

__annotations__ = {}

abstractmethod get_authorization(http_method: str, resource_uri: ParseResult, request_headers: Sequence[RequestHeader] | None) → str

Returns a value that can be used for the “Authorization” header.

Parameters:

• http_method – The HTTP method.

• resource_uri – The URI of the resource.

• request_headers – A sequence of RequestHeaders. This sequence may not be modified and may not contain headers with the same name.

class worldline.connect.sdk.authentication.authorization_type.AuthorizationType

Bases: object

AUTHORIZATION_TYPES = ['v1HMAC']

V1HMAC = 'v1HMAC'

static get_authorization(name: str) → str

class worldline.connect.sdk.authentication.v1hmac_authenticator.V1HMACAuthenticator(api_key_id: str, secret_api_key: str)

Bases: Authenticator

Authenticator implementation using v1HMAC signatures.

__abstractmethods__ = frozenset({})

__annotations__ = {}

__init__(api_key_id: str, secret_api_key: str)

Parameters:

• api_key_id – An identifier for the secret API key. The api_key_id can be retrieved from the Configuration Center. This identifier is visible in the HTTP request and is also used to identify the correct account.

• secret_api_key – A shared secret. The shared secret can be retrieved from the Configuration Center. An api_key_id and secret_api_key always go hand-in-hand, the difference is that secret_api_key is never visible in the HTTP request. This secret is used as input for the HMAC algorithm.

create_authentication_signature(data_to_sign: str) → str

get_authorization(http_method: str, resource_uri: ParseResult, http_headers: Sequence[RequestHeader] | None) → str

Returns a v1HMAC authentication signature header

static to_canonicalize_header_value(original_value: str | None) → str

to_data_to_sign(http_method: str, resource_uri: ParseResult, http_headers: Sequence[RequestHeader] | None) → str

exception worldline.connect.sdk.communication.communication_exception.CommunicationException(exception: Exception)

Bases: RuntimeError

Indicates an exception regarding the communication with the Worldline Global Collect platform such as a connection exception.

class worldline.connect.sdk.communication.connection.Connection

Bases: LoggingCapable, ObfuscationCapable, ABC

Represents a connection to the Worldline Global Collect platform server.

__abstractmethods__ = frozenset({'delete', 'disable_logging', 'enable_logging', 'get', 'post', 'put', 'set_body_obfuscator', 'set_header_obfuscator'})

__annotations__ = {}

close() → None

Releases any system resources associated with this object.

abstractmethod delete(url: str | ParseResult, request_headers: Sequence[RequestHeader]) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Send a DELETE request to the Worldline Global Collect platform and return the response.

Parameters:

• url – The URI to call, including any necessary query parameters.

• request_headers – An optional sequence of request headers.

Returns:

The response from the Worldline Global Collect platform as a tuple with the status code, headers and a generator of body chunks

Raises:

CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

abstractmethod get(url: str | ParseResult, request_headers: Sequence[RequestHeader]) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Send a GET request to the Worldline Global Collect platform and return the response.

Parameters:

• url – The URI to call, including any necessary query parameters.

• request_headers – An optional sequence of request headers.

Returns:

The response from the Worldline Global Collect platform as a tuple with the status code, headers and a generator of body chunks

Raises:

CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

abstractmethod post(url: str | ParseResult, request_headers: Sequence[RequestHeader], body: str | MultipartFormDataObject | None) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Send a POST request to the Worldline Global Collect platform and return the response.

Parameters:

• url – The URI to call, including any necessary query parameters.

• request_headers – An optional sequence of request headers.

• body – The optional body to send.

Returns:

The response from the Worldline Global Collect platform as a tuple with the status code, headers and a generator of body chunks

Raises:

CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

abstractmethod put(url: str | ParseResult, request_headers: Sequence[RequestHeader], body: str | MultipartFormDataObject | None) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Send a PUT request to the Worldline Global Collect platform and return the response.

Parameters:

• url – The URI to call, including any necessary query parameters.

• request_headers – An optional sequence of request headers.

• body – The optional body to send.

Returns:

The response from the Worldline Global Collect platform as a tuple with the status code, headers and a generator of body chunks

Raises:

CommunicationException – when an exception occurred communicating with the Worldline Global Collect platform

class worldline.connect.sdk.communication.default_connection.DefaultConnection(connect_timeout: int, socket_timeout: int, max_connections: int = 10, proxy_configuration: ProxyConfiguration | None = None)

Bases: PooledConnection

Provides an HTTP request interface, thread-safe

Parameters:

• connect_timeout – timeout in seconds before a pending connection is dropped

• socket_timeout – timeout in seconds before dropping an established connection. This is the time the server is allowed for a response

• max_connections – the maximum number of connections in the connection pool

• proxy_configuration – ProxyConfiguration object that contains data about proxy settings if present. It should be writeable as string and have a scheme attribute.

Use the methods get, delete, post and put to perform the corresponding HTTP request. Alternatively you can use request with the request method as the first parameter.

URI, headers and body should be given on a per-request basis.

__abstractmethods__ = frozenset({})

__annotations__ = {}

__enter__()

__exit__(exc_type, exc_val, exc_tb)

close() → None

Explicitly closes the connection

close_expired_connections() → None

Closes all expired HTTP connections.

close_idle_connections(idle_time: timedelta) → None

Parameters:

idle_time – a datetime.timedelta object indicating the idle time

property connect_timeout: int | None

Connection timeout in seconds

delete(url: str | ParseResult, request_headers: Sequence[RequestHeader]) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Perform a request to the server given by url

Parameters:

• url – the url to the server, given as a parsed url

• request_headers – a sequence containing RequestHeader objects representing the request headers

disable_logging() → None

Turns off logging.

enable_logging(communicator_logger: CommunicatorLogger) → None

Turns on logging using the given communicator logger.

Raises:

ValueError – If the given communicator logger is None.

get(url: str | ParseResult, request_headers: Sequence[RequestHeader]) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Perform a request to the server given by url

Parameters:

• url – the url to the server, given as a parsed url

• request_headers – a sequence containing RequestHeader objects representing the request headers

post(url: str | ParseResult, request_headers: Sequence[RequestHeader], body: str | MultipartFormDataObject | None) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Perform a request to the server given by url

Parameters:

• url – the url to the server, given as a parsed url

• request_headers – a sequence containing RequestHeader objects representing the request headers

• body – the request body

put(url: str | ParseResult, request_headers: Sequence[RequestHeader], body: str | MultipartFormDataObject | None) → Tuple[int, Mapping[str, str], Iterable[bytes]]

Perform a request to the server given by url

Parameters:

• url – the url to the server, given as a parsed url

• request_headers – a sequence containing RequestHeader objects representing the request headers

• body – the request body

set_body_obfuscator(body_obfuscator: BodyObfuscator) → None

Sets the current body obfuscator to use.

set_header_obfuscator(header_obfuscator: HeaderObfuscator) → None

Sets the current header obfuscator to use.

property socket_timeout: int | None

Socket timeout in seconds

class worldline.connect.sdk.communication.metadata_provider.MetadataProvider(integrator: str | None, shopping_cart_extension: ShoppingCartExtension | None = None, additional_request_headers: Sequence[RequestHeader] | None = ())

Bases: object

Provides meta info about the server.

__annotations__ = {'_MetadataProvider__metadata_headers': typing.Sequence[worldline.connect.sdk.communication.request_header.RequestHeader]}

property metadata_headers: Sequence[RequestHeader]

Returns:

The server related headers containing the metadata to be associated with the request (if any). This will always contain at least an automatically generated header X-GCS-ServerMetaInfo.

prohibited_headers = ('Authorization', 'Content-Type', 'Date', 'X-GCS-Idempotence-Key', 'X-GCS-ServerMetaInfo')

class worldline.connect.sdk.communication.multipart_form_data_object.MultipartFormDataObject

Bases: object

A representation of a multipart/form-data object.

add_file(parameter_name: str, uploadable_file: UploadableFile) → None

add_value(parameter_name: str, value: str) → None

property boundary: str

property content_type: str

property files: Mapping[str, UploadableFile]

property values: Mapping[str, str]

class worldline.connect.sdk.communication.multipart_form_data_request.MultipartFormDataRequest

Bases: ABC

A representation of a multipart/form-data request.

__abstractmethods__ = frozenset({'to_multipart_form_data_object'})

__annotations__ = {}

abstractmethod to_multipart_form_data_object() → MultipartFormDataObject

Returns:

worldline.connect.sdk.communication.MultipartFormDataObject

exception worldline.connect.sdk.communication.not_found_exception.NotFoundException(exception: Exception, message: str)

Bases: RuntimeError

Indicates an exception that occurs when the requested resource is not found. In normal usage of the SDK, this exception should not occur, however it is possible. For example when path parameters are set with invalid values.

class worldline.connect.sdk.communication.param_request.ParamRequest

Bases: ABC

Represents a set of request parameters.

__abstractmethods__ = frozenset({'to_request_parameters'})

__annotations__ = {}

abstractmethod to_request_parameters() → List[RequestParam]

Returns:

list[worldline.connect.sdk.communication.RequestParam] representing the HTTP request parameters

class worldline.connect.sdk.communication.pooled_connection.PooledConnection

Bases: Connection, ABC

Represents a pooled connection to the Worldline Global Collect platform server. Instead of setting up a new HTTP connection for each request, this connection uses a pool of HTTP connections.

__abstractmethods__ = frozenset({'close_expired_connections', 'close_idle_connections', 'delete', 'disable_logging', 'enable_logging', 'get', 'post', 'put', 'set_body_obfuscator', 'set_header_obfuscator'})

__annotations__ = {}

abstractmethod close_expired_connections() → None

Closes all expired HTTP connections.

abstractmethod close_idle_connections(idle_time: timedelta) → None

Closes all HTTP connections that have been idle for the specified time. This should also include all expired HTTP connections.

Parameters:

idle_time – a datetime.timedelta object indicating the idle time

class worldline.connect.sdk.communication.request_header.RequestHeader(name: str, value: str | None)

Bases: object

A single request header. Immutable.

property name: str

Returns:

The header name.

property value: str | None

Returns:

The un-encoded value.

worldline.connect.sdk.communication.request_header.get_header(headers: Sequence[RequestHeader] | Mapping[str, str] | None, header_name: str) → RequestHeader | None

Returns:

The header with the given name, or None if there was no such header.

worldline.connect.sdk.communication.request_header.get_header_value(headers: Sequence[RequestHeader] | Mapping[str, str] | None, header_name: str) → str | None

Returns:

The value of the header with the given name, or None if there was no such header.

class worldline.connect.sdk.communication.request_param.RequestParam(name: str, value: str | None)

Bases: object

A single request parameter. Immutable.

property name: str

Returns:

The parameter name.

property value: str | None

Returns:

The un-encoded value.

exception worldline.connect.sdk.communication.response_exception.ResponseException(status: int, body: str | None, headers: Mapping[str, str] | None)

Bases: RuntimeError

Thrown when a response was received from the Worldline Global Collect platform which indicates an error.

property body: str | None

Returns:

The raw response body that was returned by the Worldline Global Collect platform.

get_header(header_name: str) → Tuple[str, str] | None

Returns:

The header with the given name, or None if there was no such header.

get_header_value(header_name: str) → str | None

Returns:

The value header with the given name, or None if there was no such header.

property headers: Mapping[str, str]

Returns:

The headers that were returned by the Worldline Global Collect platform. Never None.

property status_code: int

Returns:

The HTTP status code that was returned by the Worldline Global Collect platform.

worldline.connect.sdk.communication.response_header.get_disposition_filename(headers: Mapping[str, str] | None) → str | None

Returns:

The value of the filename parameter of the Content-Disposition header, or None if there was no such header or parameter.

worldline.connect.sdk.communication.response_header.get_header(headers: Mapping[str, str] | None, header_name: str) → Tuple[str, str] | None

Returns:

The header with the given name as a tuple with the name and value, or None if there was no such header.

worldline.connect.sdk.communication.response_header.get_header_value(headers: Mapping[str, str] | None, header_name: str) → str | None

Returns:

The value of the header with the given name, or None if there was no such header.

class worldline.connect.sdk.domain.data_object.DataObject

Bases: object

from_dictionary(dictionary: dict) → DataObject

to_dictionary() → dict

class worldline.connect.sdk.domain.shopping_cart_extension.ShoppingCartExtension(creator: str, name: str, version: str, extension_id: str | None = None)

Bases: DataObject

__annotations__ = {}

static create_from_dictionary(dictionary: dict) → ShoppingCartExtension

property creator: str

property extension_id: str | None

from_dictionary(dictionary: dict) → ShoppingCartExtension

property name: str

to_dictionary() → dict

property version: str

class worldline.connect.sdk.domain.uploadable_file.UploadableFile(file_name: str, content: Any, content_type: str, content_length: int = -1)

Bases: object

A file that can be uploaded.

The allowed forms of content are defined by the Connection implementation. The default implementation supports strings, file descriptors and io.BytesIO objects.

property content: Any

Returns:

The file’s content.

property content_length: int

Returns:

The file’s content length, or -1 if not known.

property content_type: str

Returns:

The file’s content type.

property file_name: str

Returns:

The name of the file.

class worldline.connect.sdk.json.default_marshaller.DefaultMarshaller

Bases: Marshaller

Marshaller implementation based on json.

__abstractmethods__ = frozenset({})

__annotations__ = {}

static instance() → DefaultMarshaller

marshal(request_object: Any) → str

Marshal a request object to a JSON string.

Parameters:

request_object – the object to marshal into a serialized JSON string

Returns:

the serialized JSON string of the request_object

unmarshal(response_json: str | bytes | None, type_class: Type[T]) → T

Unmarshal a JSON string to a response object.

Parameters:

• response_json – the json body that should be unmarshalled

• type_class – The class to which the response_json should be unmarshalled

Raises:

MarshallerSyntaxException – if the JSON is not a valid representation for an object of the given type

class worldline.connect.sdk.json.marshaller.Marshaller

Bases: ABC

Used to marshal and unmarshal Worldline Global Collect platform request and response objects to and from JSON.

__abstractmethods__ = frozenset({'marshal', 'unmarshal'})

__annotations__ = {}

abstractmethod marshal(request_object: Any) → str

Marshal a request object to a JSON string.

Parameters:

request_object – the object to marshal into a serialized JSON string

Returns:

the serialized JSON string of the request_object

abstractmethod unmarshal(response_json: str | bytes | None, type_class: Type[T]) → T | None

Unmarshal a JSON string to a response object.

Parameters:

• response_json – the json body that should be unmarshalled

• type_class – The class to which the response_json should be unmarshalled

Raises:

MarshallerSyntaxException – if the JSON is not a valid representation for an object of the given type

exception worldline.connect.sdk.json.marshaller_syntax_exception.MarshallerSyntaxException(cause: Exception | None = None)

Bases: RuntimeError

Thrown when a JSON string cannot be converted to a response object.

class worldline.connect.sdk.log.body_obfuscator.BodyObfuscator(additional_rules: Mapping[str, Callable[[str], str]] | None = None)

Bases: object

A class that can be used to obfuscate properties in JSON bodies.

__annotations__ = {'_BodyObfuscator__obfuscation_rules': typing.Dict[str, typing.Callable[[str], str]], '_BodyObfuscator__property_pattern': typing.Pattern[~AnyStr]}

__init__(additional_rules: Mapping[str, Callable[[str], str]] | None = None)

Creates a new body obfuscator. This will contain some pre-defined obfuscation rules, as well as any provided custom rules

Parameters:

additional_rules – An optional mapping from property names to obfuscation rules, where an obfuscation rule is a function that obfuscates a single string,

static default_body_obfuscator() → BodyObfuscator

obfuscate_body(body: str | bytes | None, charset: str | None = None) → str | None

Obfuscates the body from the given stream as necessary. :param body: The body to obfuscate, as string or bytes. :param charset: The charset to use to read the body bytes.

class worldline.connect.sdk.log.communicator_logger.CommunicatorLogger

Bases: ABC

Used to log messages from communicators.

__abstractmethods__ = frozenset({'log'})

__annotations__ = {}

abstractmethod log(message: str, thrown: Exception | None = None) → None

Logs a throwable with an accompanying message.

Parameters:

• message – The message accompanying the throwable.

• thrown – The throwable to log.

log_request(request_log_message: RequestLogMessage) → None

Logs a request message object

log_response(response_log_message: ResponseLogMessage) → None

Logs a response message object

class worldline.connect.sdk.log.header_obfuscator.HeaderObfuscator(additional_rules: Mapping[str, Callable[[str], str]] | None = None)

Bases: object

A class that can be used to obfuscate headers.

__annotations__ = {'_HeaderObfuscator__obfuscation_rules': typing.Dict[str, typing.Callable[[str], str]]}

__init__(additional_rules: Mapping[str, Callable[[str], str]] | None = None)

Creates a new header obfuscator. This will contain some pre-defined obfuscation rules, as well as any provided custom rules

Parameters:

additional_rules – An optional mapping from property names to obfuscation rules, where an obfuscation rule is a function that obfuscates a single string,

static default_header_obfuscator() → HeaderObfuscator

obfuscate_header(header_name: str, value: str) → str

Obfuscates the value for the given header as necessary.

class worldline.connect.sdk.log.log_message.LogMessage(request_id: str, body_obfuscator: BodyObfuscator = <worldline.connect.sdk.log.body_obfuscator.BodyObfuscator object>, header_obfuscator: HeaderObfuscator = <worldline.connect.sdk.log.header_obfuscator.HeaderObfuscator object>)

Bases: ABC

A utility class to build log messages.

__abstractmethods__ = frozenset({'get_message'})

__annotations__ = {'_LogMessage__body': typing.Optional[str], '_LogMessage__body_obfuscator': <class 'worldline.connect.sdk.log.body_obfuscator.BodyObfuscator'>, '_LogMessage__content_type': typing.Optional[str], '_LogMessage__header_list': typing.List[typing.Tuple[str, str]], '_LogMessage__header_obfuscator': <class 'worldline.connect.sdk.log.header_obfuscator.HeaderObfuscator'>, '_LogMessage__headers': <class 'str'>, '_LogMessage__request_id': <class 'str'>, '__body': 'Optional[str]', '__body_obfuscator': 'BodyObfuscator', '__content_type': 'Optional[str]', '__header_list': 'List[Tuple[str, str]]', '__header_obfuscator': 'HeaderObfuscator', '__headers': 'str', '__request_id': 'str'}

add_header(name: str, value: str | None) → None

property body: str | None

property content_type: str | None

static empty_if_none(value: str | None) → str

get_header_list() → Sequence[Tuple[str, str]]

abstractmethod get_message() → str

property headers: str

property request_id: str

set_binary_body(content_type: str | None) → None

set_body(body: str | bytes | None, content_type: str | None, charset: str | None = None) → None

class worldline.connect.sdk.log.logging_capable.LoggingCapable

Bases: ABC

Classes that extend this class have support for log messages from communicators.

__abstractmethods__ = frozenset({'disable_logging', 'enable_logging'})

__annotations__ = {}

abstractmethod disable_logging() → None

Turns off logging.

abstractmethod enable_logging(communicator_logger: CommunicatorLogger) → None

Turns on logging using the given communicator logger.

Raises:

ValueError – If the given communicator logger is None.

class worldline.connect.sdk.log.obfuscation_capable.ObfuscationCapable

Bases: ABC

Classes that extend this class support obfuscating bodies and headers.

__abstractmethods__ = frozenset({'set_body_obfuscator', 'set_header_obfuscator'})

__annotations__ = {}

abstractmethod set_body_obfuscator(body_obfuscator: BodyObfuscator) → None

Sets the current body obfuscator to use.

abstractmethod set_header_obfuscator(header_obfuscator: HeaderObfuscator) → None

Sets the current header obfuscator to use.

worldline.connect.sdk.log.obfuscation_rule.obfuscate_all() → Callable[[str], str]

Returns an obfuscation rule (function) that will replace all characters with *

worldline.connect.sdk.log.obfuscation_rule.obfuscate_all_but_first(count: int) → Callable[[str], str]

Returns an obfuscation rule (function) that will keep a fixed number of characters at the start, then replaces all other characters with *

worldline.connect.sdk.log.obfuscation_rule.obfuscate_all_but_last(count: int) → Callable[[str], str]

Returns an obfuscation rule that will keep a fixed number of characters at the end, then replaces all other characters with *

worldline.connect.sdk.log.obfuscation_rule.obfuscate_with_fixed_length(fixed_length: int) → Callable[[str], str]

Returns an obfuscation rule (function) that will replace values with a fixed length string containing only *

class worldline.connect.sdk.log.python_communicator_logger.PythonCommunicatorLogger(logger: Logger, log_level: int, error_log_level: int | None = None)

Bases: CommunicatorLogger

A communicator logger that is backed by the log library.

__abstractmethods__ = frozenset({})

__annotations__ = {}

__init__(logger: Logger, log_level: int, error_log_level: int | None = None)

Logs messages to the argument logger using the argument log_level. If absent, the error_log_level will be equal to the log_level. Note that if the CommunicatorLogger’s log level is lower than the argument logger’s log level (e.g. the CommunicatorLogger is given log.INFO as level and the argument logger has a level of log.WARNING), then nothing will be logged to the logger.

Parameters:

• logger – the logger to log to

• log_level – the log level that will be used for non-error messages logged via the CommunicatorLogger

• error_log_level – the log level that will be used for error messages logged via the CommunicatorLogger.

log(message: str, thrown: Exception | None = None) → None

Log a message to the underlying logger. If thrown is absent, the message will be logged with the CommunicatorLogger’s log_level, if a thrown object is provided, the message and exception will be logged with the CommunicatorLogger’s error_log_level.

Parameters:

• message – the message to be logged

• thrown – an optional throwable object

class worldline.connect.sdk.log.request_log_message.RequestLogMessage(request_id: str, method: str, uri: str, body_obfuscator: BodyObfuscator = <worldline.connect.sdk.log.body_obfuscator.BodyObfuscator object>, header_obfuscator: HeaderObfuscator = <worldline.connect.sdk.log.header_obfuscator.HeaderObfuscator object>)

Bases: LogMessage

A utility class to build request log messages.

__abstractmethods__ = frozenset({})

__annotations__ = {}

get_message() → str

class worldline.connect.sdk.log.response_log_message.ResponseLogMessage(request_id: str, status_code: int, duration: int = -1, body_obfuscator: BodyObfuscator = <worldline.connect.sdk.log.body_obfuscator.BodyObfuscator object>, header_obfuscator: HeaderObfuscator = <worldline.connect.sdk.log.header_obfuscator.HeaderObfuscator object>)

Bases: LogMessage

A utility class to build request log messages.

__abstractmethods__ = frozenset({})

__annotations__ = {}

get_duration() → int

get_message() → str

get_status_code() → int

class worldline.connect.sdk.log.sys_out_communicator_logger.SysOutCommunicatorLogger

Bases: CommunicatorLogger

A communicator logger that prints its message to sys.stdout It includes a timestamp in yyyy-MM-ddTHH:mm:ss format in the system time zone.

__abstractmethods__ = frozenset({})

__annotations__ = {}

static instance() → SysOutCommunicatorLogger

log(message: str, thrown: Exception | None = None) → None

Logs a throwable with an accompanying message.

Parameters:

• message – The message accompanying the throwable.

• thrown – The throwable to log.