Skip to content

Instantly share code, notes, and snippets.

@jakebrinkmann
Created June 6, 2024 01:24
Show Gist options
  • Save jakebrinkmann/064f384fc61d76ec4a3ac40936012a30 to your computer and use it in GitHub Desktop.
Save jakebrinkmann/064f384fc61d76ec4a3ac40936012a30 to your computer and use it in GitHub Desktop.
Example using suds library to make SOAP requests in python
from soap_client import SoapClient

client = SoapClient("<host>", "<username>", "<password>")
client.doOperation()

options = {}
client.doThingWithParams(options)
import logging
import time
from datetime import datetime, timezone
import suds.client
from suds.client import WebFault
logger = logging.getLogger()
class ElapsedTimeTracker:
def __init__(self):
self.start_time = time.time()
def elapsed(self):
duration = datetime.fromtimestamp(time.time() - self.start_time, timezone.utc)
return duration.strftime("%H:%M:%S")
class SoapClient:
"""Performs interactions with a SOAP API"""
def __init__(
self, wsdl_host: str, wsdl_username: str, wsdl_password: str, timeout=90
):
self.wsdl_host = wsdl_host
self.connection = suds.client.Client(
self.wsdl_host,
soapheaders={
"credentials": {
"username": wsdl_username,
"password": wsdl_password,
}
},
timeout=timeout,
)
self.connection.set_options(location=wsdl_host)
def __getattr__(self, name):
def handler(*args, **kwargs):
return self.__action(name, *args, **kwargs)
return handler
@property
def last_sent(self):
return self.connection.last_sent().str()
@property
def last_received(self):
return self.connection.last_received().str()
def __action(self, action: str, params: dict = {}):
success = False
response = None
__timer = ElapsedTimeTracker()
try:
response = getattr(self.connection.service, action)(**params)
success = True
except WebFault as exc:
response = str(exc.fault.faultstring)
raise
except TimeoutError as exc:
response = str(exc)
raise
finally:
# print("Last Sent ==> \n\n" + self.last_sent)
# print("Last Received ==> \n\n" + self.last_received)
logger.info(
{
"host": self.wsdl_host,
"action": action,
"duration": __timer.elapsed(),
"success": success,
"response": str(response).replace("\n", " "),
}
)
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment