How to use the zeep.Client function in zeep

Test service
            
              
            
          
        
    """.strip()
    )

    transport = DummyTransport()
    client = Client(wsdl_main, transport=transport, plugins=[wsa.WsAddressingPlugin()])
    binding = client.wsdl.services.get("TestService").ports.get("TestPortType").binding

    envelope, headers = binding._create(
        "TestOperation1",
        args=["foo"],
        kwargs={},
        client=client,
        options={"address": "http://tests.python-zeep.org/test"},
    )
    expected = """
        
          
            urn:dummyRequest
            urn:uuid:ada686f9-5995-4088-bea4-239f694b2eaf
            http://tests.python-zeep.org/test

def request_transaction(self, transaction: Transaction) -> Transaction:
        client = Client(self._server_url)
        if 'proxies' in self.config:
            client.transport.session.proxies = self.config['proxies']
        try:
            params = {
                'merchantConfigurationID': self.config['merchant_config_id'],
                'encryptedRequest': self._get_encrypted_request(transaction)
            }

            result = client.service.RequestOperation(**params)
            if not result:
                raise TransactionError('AsanPardakht: invalid information.')

            if result[0:1] == '0':
                exploded_result = str(result).split(',')  # Should have two part [XX,YY]
                transaction.id = exploded_result[1]
            else:

self.verify_cert = resource_filename('bankid.certs', 'appapi.test.bankid.com.pem')
        else:
            self.api_url = 'https://appapi.bankid.com/rp/v4'
            self.wsdl_url = 'https://appapi.bankid.com/rp/v4?wsdl'
            self.verify_cert = resource_filename('bankid.certs', 'appapi.bankid.com.pem')

        headers = {
            "Content-Type": "text/xml;charset=UTF-8",
        }

        session = requests.Session()
        session.verify = self.verify_cert
        session.cert = self.certs
        session.headers = headers
        transport = Transport(session=session)
        self.client = Client(self.wsdl_url, transport=transport)

"""Get a SOAP client for performing requests. The client is cached. The
    timeout is in seconds."""
    # this function isn't automatically tested because the functions using
    # it are not automatically tested
    if (wsdlurl, timeout) not in _soap_clients:
        # try zeep first
        try:
            from zeep.transports import Transport
            transport = Transport(timeout=timeout)
            from zeep import CachingClient
            client = CachingClient(wsdlurl, transport=transport).service
        except ImportError:
            # fall back to non-caching zeep client
            try:
                from zeep import Client
                client = Client(wsdlurl, transport=transport).service
            except ImportError:
                # other implementations require passing the proxy config
                try:
                    from urllib import getproxies
                except ImportError:
                    from urllib.request import getproxies
                # fall back to suds
                try:
                    from suds.client import Client
                    client = Client(
                        wsdlurl, proxy=getproxies(), timeout=timeout).service
                except ImportError:
                    # use pysimplesoap as last resort
                    try:
                        from pysimplesoap.client import SoapClient
                        client = SoapClient(

from __future__ import print_function

from requests import Session
from requests.auth import HTTPBasicAuth

import zeep
from zeep.transports import Transport

# Example using basic authentication with a webservice

session = Session()
session.auth = HTTPBasicAuth('username', 'password')
transport_with_basic_auth = Transport(session=session)

client = zeep.Client(
    wsdl='http://nonexistent?WSDL',
    transport=transport_with_basic_auth
)

client.wsdl.dump()

for elem in soup.find_all(class_="id"):
				sitecodes.append(elem.get_text())

		if not hasattr(parent, 'poll_shc_soap_deltat'):
			logger.error("poll_shc_soap_deltat must be specified to form soap query")
			return
		deltat = parent.poll_shc_soap_deltat[0]		

		# Form mindate and maxdate part of soap query, with the deltat param provided
		run_time = datetime.datetime.utcnow()
		run_time_file = run_time.strftime('%Y%m%d_%H%M')
		fetch_date_end = run_time
		fetch_date_start = fetch_date_end - datetime.timedelta(hours=int(deltat))
		end = fetch_date_end.strftime('%Y-%m-%d %H:%M:%S')
		start = fetch_date_start.strftime('%Y-%m-%d %H:%M:%S')
		client = zeep.Client('https://ws-shc.qc.dfo-mpo.gc.ca/observations?wsdl')

		for site in sitecodes:
			# from the wdsl page: 
			wdata = client.service.search('wl',40.0,85.0,-145.0,-50.0,0.0,0.0,start,end,1,1000,'true','station_id='+site,'asc')
			if wdata['status']['status'] == 'ok':
				# new_baseurl contains the query parameters to download the file. It messes up
				# sarracenia to transmit a new_baseurl with spaces, so I took the spaces out of
				# the dates to transmit
				logger.info("poll_shc_soap file ready %s" % \
						'shc_{0}_{1}.dat'.format(run_time_file,site))
				mstart = datetime.datetime.strptime(start,'%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%d_%H:%M:%S')
				mend = datetime.datetime.strptime(end,'%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%d_%H:%M:%S')
				parent.msg.new_baseurl = """wl,40.0,85.0,-145.0,-50.0,0.0,0.0,{0},{1},1,1000,true,station_id={2},asc""".format(mstart,mend,site)
				parent.msg.new_file = 'shc_{0}_{1}.dat'.format(run_time_file,site)

self.service_name = service_name
        self.mobile_message = mobile_message

        self.session_code = None
        self.data_files = []
        self.container = None

        if wsdl_url == 'https://tsp.demo.sk.ee/dds.wsdl':  # pragma: no branch
            assert service_name == 'Testimine', 'When using Test DigidocService the service name must be `Testimine`'

        the_transport = transport or Transport(cache=SqliteCache())

        if ZeepSettings is not None:  # pragma: no branch
            settings = ZeepSettings(strict=False)

            self.client = Client(wsdl_url, transport=the_transport, settings=settings)

        else:
            self.client = Client(wsdl_url, transport=the_transport, strict=False)

def _get_zeep_client(self, transport_to_bind=None):
        if not transport_to_bind:
            raise HPNAConnectionError("SOAP client cannot be instantiated without a transport.")
        zeep_client = ZeepClient(wsdl=self._wsdl_url, transport=transport_to_bind)
        return zeep_client

def create_mandate_bse(client_code, amount):
	'''
	Creates mandate for user for a specific amount 
	Called before creating an SIP transaction on BSEStar because 
		every SIP transaction requires a mandate
	'''

	## initialise the zeep client for wsdl
	client = zeep.Client(wsdl=WSDL_UPLOAD_URL[settings.LIVE])
	set_soap_logging()

	## get the password
	pass_dict = soap_get_password_upload(client)
	
	## prepare the mandate record 
	bse_mandate = prepare_mandate_param(client_code, amount)

	## post the mandate creation request
	mandate_id = soap_create_mandate(client, bse_mandate, pass_dict)
	return mandate_id