Custom Soap Header if element does not have a namespace?

[johnziebro]

The API I am working with requires a custom defined api_key in the soap header. Successful authentication with the following using Sud's Custom Soap Headers docs:

# create auth soap_header
api_key = "abc123"
auth_ns = ('apiKey', 'https://api.redacted.net/2.0/')
api_key = Element('api_key', ns=auth_ns).setText(api_key)
self.client.set_options(soapheaders=api_key)

The API also uses a pager for multiple records defined in the soap header. I am having issues getting it to work in parallel with authentication. Note that the pager and auth are not defined in the WSDL. Also, the pager does not have a namespace (ns) like the auth does.

Documentation Provided:

  <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
  xmlns:ns="https://api.redacted.net/2.0/">
     <soapenv:Header>
       <ns:apiKey>
          <api_key>***</api_key>
       </ns:apiKey>
       <pager>
          <page>2</page>
          <per_page>100</per_page>
        </pager>
     </soapenv:Header>
     <soapenv:Body>
        <ns:getCaseUpdates>
           <case_id>2038217180</case_id>
        </ns:getCaseUpdates>
     </soapenv:Body>
  </soapenv:Envelope>

This however does not work when including the pager:

# create auth soap_header
api_key = "abc123"
auth_ns = ('apiKey', 'https://api.redacted.net/2.0/')
auth = Element('api_key', ns=auth_ns).setText(CFG.get(f"{self.config_prefix}KEY"))
#self.client.set_options(soapheaders=ns)

# create pager
pager_ns = ('pager') # no namespace, see spec above
page = Element('page', ns=pager_ns).setText(2)
per_page = Element('per_page', ns=pager_ns).setText(50)

# set client options with headers
self.client.set_options(soapheaders=(auth, (page, per_page)))

Providing an empty string or None from pager_ns namespace also fails. If I could get this to work for an api call, I could abstract the pager out for use only when needed. Any help would be appreciated. Thank you.

[phillbaker]

Hello, would something like the following work?

>>> from suds.sax.element import Element
>>> pager = Element('pager')
>>> page = Element('page').setText(2)
>>> per_page = Element('per_page').setText(50)
>>> pager.append(page)
>>> pager.append(per_page)
>>> print(pager)
<pager>
   <page>2</page>
   <per_page>50</per_page>
</pager>
>>> self.client.set_options(soapheaders=(auth, pager))

Please note: contributors not generally able to provide support for using suds, other support oriented sites like stackoverflow may be able to help and contain answers to common questions. If you determine that there are specific features or bugs you encounter while working on this, please let us know.

In order to reproduce issues, please provide:

• Package and version of suds being used
• Generated output XML from loggers
• A publicly available url for the wsdl in question and a runnable code snippet OR a test case added to the test suite demonstrating the issue

[johnziebro]

Will test tomorrow and report back. Thank you for the help.

[johnziebro]

Working on this today. Before I start I just wanted to say thanks. I've worked with Zeep, and have found that simply getting an answer from the community or maintainer highly problematic. The question I asked above I've had outstanding on SO for quite some time with Zeep, as well as posted in the Github issues.

Thank you for being responsive even though it adds extra work. If I get the above working I plan on migrating over to Suds.

[johnziebro]

@phillbaker, this solution worked beautifully. Thank you again, I am moving over to Suds from Zeep. Including relevant notes for those who might ask a similar question later.

In my implementation I use a base class that implements the Suds client. This is for reuse purposes, if I need to work with other SOAP APIs I already have the basics in place.

In my base class I handled the paging this way:

    @rate_limit
    def call_(self, endpoint: str, *args, **kwargs):
        """Allows calling of any client service defined in the WSDL by name."""
        try:

            paging = False

            # extract page and per_page if in kwargs
            if "page" in kwargs:
                paging = True
                pager_settings = {"page": kwargs.pop("page")}

                if "per_page" in kwargs:
                    pager_settings["per_page"] = kwargs.pop("per_page")

                # set client options to include auth and configured pager element
                self.soap_client.set_options(soapheaders=(self.__get_api_key(), self.__get_pager(**pager_settings)))

            # get the endpoint
            endpoint = self.get_endpoint(endpoint)

            # call the endpoint with provided unnamed and named parameters if any
            result = endpoint(*args, **kwargs)

            # remove pager element from header in case next call does not include it
            if paging:
                self.soap_client.set_options(soapheaders=self.__get_api_key())

        except Exception as error:
            local_vars = locals()
            for keys in ['error', 'self', 'headers']:
                local_vars.pop(keys, None)  # remove headers, specifically api_key to prevent security leakage to logs
            error = f"{str(error)} - {local_vars}"
            log.exception(error)
            return None

        else:
            return result

    def get_endpoint(self, name: str):
        """ Allows accessing any service by name. """
        return getattr(self.soap_client.service, name)

    def __get_pager(self, page: int = 1, per_page: int = 100):
        """ Create non-namespaced pager element that contains the page and records per page. """
        pager = Element('pager')
        page = Element('page').setText(page)
        per_page = Element('per_page').setText(per_page)
        pager.append(page)
        pager.append(per_page)
        return pager

    def __get_api_key(self):
        """ Create header element that contains the api key. """
        api_key = CFG.get(f"{self.get_config_prefix()}KEY")
        auth_ns = ('apiKey', 'https://api.redacted.net/2.0/')
        return Element('api_key', ns=auth_ns).setText(api_key)

In Zeep, soapheaders can be passed when calling an endpoint, but I haven't investigated that, session caching or transports yet. If soapheaders can be passed with the individual requests, it might simplify the paging mechanics instead of updating them on the instantiated client which I store as an attribute of the base class.