Python-zeep: Impossible to initialize async instance if loop currently running

[penguinolog]

1. The version of zeep (or if you are running master the commit hash/date)
   zeep==1.1.0
2. The WSDL you are using
   https://webservices2.autotask.net/ATServices/1.5/atws.asmx?WSDL

3. An example script

   Without sensetive data

   import asyncio

   loop = asyncio.get_event_loop()

   async def func():
   def start_client():
   session = aiohttp.ClientSession(
   auth=aiohttp.BasicAuth(
   login='login',
   password='password'
   ),
   loop=loop
   ) # loop is running -> no warnings

       transport = zeep.asyncio.AsyncTransport(
           loop=loop,
           cache=zeep.cache.InMemoryCache(),
           session=session
       )
   
   client = zeep.Client(
       wsdl=wsdl,
       transport=transport
   )
   

Caused:

lib/python3.5/site-packages/zeep/asyncio/transport.py", line 45, in _load_remote_data
    self.loop.run_until_complete(_load_remote_data_async())
  File "uvloop/loop.pyx", line 1198, in uvloop.loop.Loop.run_until_complete (uvloop/loop.c:25480)
  File "uvloop/loop.pyx", line 1191, in uvloop.loop.Loop.run_until_complete (uvloop/loop.c:25324)
  File "uvloop/loop.pyx", line 1100, in uvloop.loop.Loop.run_forever (uvloop/loop.c:24056)
  File "uvloop/loop.pyx", line 355, in uvloop.loop.Loop._run (uvloop/loop.c:11384)
RuntimeError: this event loop is already running.

Solution:

if self.loop.is_running():
    # here we should call synchronous API (requests)
    # it may look like ugly, but currently it's only one solution
else:
    self.loop.run_until_complete(
        _load_remote_data_async()
    )

[jskolovos]

I have problem with this and I am new to Async too. Looking the source the problem is in the zeep/asyncio/transport module:

`

def _load_remote_data(self, url):
    result = None

    async def _load_remote_data_async():
        nonlocal result
        with aiohttp.Timeout(self.load_timeout):
            response = await self.session.get(url)
            result = await response.read()
            try:
                response.raise_for_status()
            except aiohttp.ClientError as exc:
                raise TransportError(
                    message=str(exc),
                    status_code=response.status,
                    content=result
                ).with_traceback(exc.__traceback__) from exc

    # Block until we have the data
    self.loop.run_until_complete(_load_remote_data_async())
    return result

`

In my understanding, that "self.loop.run_until_complete(_load_remote_data_async())" line should not go in async code, and it is causing the error here reported.

The way to "Block until we have the data" in async code, is "await".

Another problem I see with this method, is the use of the "nonlocal result" line.
All the async pending operations only put in the stack the local values of the async method, and the nonlocal or global variables, are overwriten by every call, so, if one call interrupts because long IO delay and another enters this code, the nonlocal value is overwriten and can give problems depending of the execution timing.
I think this is the reason of the:

`

    # Block until we have the data
    self.loop.run_until_complete(_load_remote_data_async())

`

By only looking this part of the code the solution is to make async the _load_remote_data(self, url) in the Async transport, and return the result from there when it finish, putting the code of the inner _load_remote_data_async() in it.

But I don't know the rest of the code and I think that @mvantellingen had a good reason to put the run_until_complete line there.

I think everybody trying to use zeep in async faces this problem or
does somebody have managed to run zeep in async mode???, please send an example.

I checked the async example provided, but the async code executes on the event_loop without any other module using the event loop.

In my case, I'm using Tornado, with asyncpg and tried to use zeep in async, but because of this problem and because time restrictions. I am using zeep synchronously, by now,

I want to thanks @mvantellingen for this project that allows me to upgrade from suds, and despite of being the only one developing actively, although not perfect, is a well crafted piece of code and well documented too.

[leric]

@jskolovos a usable work around is to initialize a Document object yourself and pass it to a client constructor, checkout my PR, it work with tornado too.

[jskolovos]

@leric, Thanks for your comment, I will try that.

[chrisimcevoy]

@jskolovos - the await keyword doesn't mean 'block until'. It represents an explicit context switch which returns control to the scheduler; By awaiting a coroutine object, we're telling the event loop (the scheduler) to get on with running/resume other Tasks until the OS flags the socket/file descriptor or what-have-you which we're awaiting as being in a readable/writeable state. Explicit context switching via await is how cooperative multitasking works in the context of asyncio, and other libraries which implement alternative schedulers/event loops.

Actually, this is the (only) reason why I was sad to see the old-style yield from syntax introduced in CPython3.4's (then-provisional) implementation of asyncio become deprecated. The semantics were much more indicative of what was actually going on under the hood. Every time you see await, think "yield control to the event loop". See this talk by Robert Smallshire which encapsulates all this brilliantly.

To reemphasise my comments under #251 and #382, all blocking io-bound operations __must__ be avoided at all costs when working with asyncio. So for me, @penguinolog's solution isn't workable. Your WSDL might be served from a slooow server for all I know, and I don't want the hundreds of concurrent aiohttp requests I'm awaiting at any given moment to be held up by a single requests.get() just to fetch it.

As we all know, you can only await a coroutine object inside of a coroutine function defined with async def, not inside a regular old synchronous def foo(): ...; function. The problem here is that def _load_remote_data() can't await it's inner async def _load_remote_data_async(). It's an awkward positioning of a synchronous request in an asynchronous context.

The current implementation may make use of an event loop and a coroutine, but it is not compatible with an asynchronous programming style. Functionally (no pun intended) there is no difference between:

self.loop.run_until_complete(_load_remote_data_async())

and:

self.session.get(url, timeout=self.load_timeout)

They're both examples of synchronous programming. To prove it, try executing multiple such calls in two separate programs. They'll take roughly the same amount of time to complete (all external factors like network latency being equal).

Currently, I tend to use a similar technique to what @leric suggested under #592 and #593, whereby I fetch a WSDL asynchronously outside of zeep and pass the result to the Client constructor. (I previously recommended that approach under #382).

The initial asyncio support which I submitted under #207 was an attempt at being fully compliant (or should I say cooperative?) with the asyncio programming paradigm. Where applicable for Async~ variants of zeep classes, I moved attribute initialisation out of __init__ and into a classmethod where coroutines could be awaited, e.g.

async def main():
    t = AsyncTransport(loop)
    c = await AsyncClient.create(wsdl, transport=t)
    ...

It was (rightly) rejected though, on the grounds that it would fracture the public API to an unacceptable degree.

I occasionally implement something similar to that PR, depending on what I need at the time. So you might find some ideas in there, although it's based on a fairly historic fork at this point.

(I don't mean to go off on a tangent or hijack the issue, but I wonder whether Document ought to be instigating external calls via Transport/AsyncTransport in the first place. It seems a little strange conceptually and in violation of the SRP...? Shouldn't Document just concern itself with being a document rather than assuming knowledge about transports and the outside world?)

[brainmorsel]

If somebody looking for workaround:

import io
import asyncio
import zeep
import zeep.exceptions
from zeep.asyncio import AsyncTransport

async def make_client(endpoint):
    transport = AsyncTransport(asyncio.get_event_loop())
    async with transport.session.get(endpoint) as resp:
        content = await resp.read()
        wsdl = io.BytesIO(content)
        return zeep.Client(wsdl, transport=transport)

This snippet allows you to create a client from running loop.

[dsod]

To use aiohttp 3.0+ aiohttp.Timeout need to be changed to async_timeout.timeout

Docs:
Drop aiohttp.Timeout and use async_timeout.timeout instead. (#2348)

https://docs.aiohttp.org/en/stable/changes.html > 3.0.0 Deprecations and Removals

[mvantellingen]

I have been looking at asyncio support again last two hours, and a few points:

The WSDL loading is not compatible with asyncio; due to the fact that http requests need to happen during the parsing of the WSDL and XSD parsing (think of xsd:import etc). This will result in making the complete codepaths async compatible. Which would result in a lot of code duplication.

Besides that there lxml can also load external files for which zeep implements a custom resolver. This might also be a blocker for truly async codepaths.

Pre-loading the WSDL content may work for the use case where the WSDL is relatively simple (doesn't contain external resources), so we should support that I think. Another option is to use run_in_executor(), but I don't have the experience to know if there are any downsides for that approach.

A completely other option is to split zeep into three code bases: zeep, aiozeep, shared-code. But I'm afraid that is a little too big for the amount of time I currently have for zeep

[chrisimcevoy]

The "other option" sounds like it might be the best way forward, @mvantellingen

All that the proposed aiozeep would need to do is provide a public API which, while being as close to standard zeep as possible, is async first rather than the halfway-house we have currently.

I'm not sure that a shared-code project would be necessary or even desirable. If there is anything to take away from the current zeep asyncio implementation, it's that mixing asyncio with non-asyncio code in the same codebase is a bit of a headache.

aiozeep just needs to be free to inherit or override zeep functionality as required.

Splitting it out would reduce the maintenance burden on you, and free you to concentrate on other things.

[dwyanepolinski]

Hello,

If you're looking for some workaround to use zeep client in async function when your main loop is running, just use aiohttp. I wrote some class to handle this. I'm not creating zeep client asynchronously, but you can try in your case.

Class for request wrapping

class ZeepAsyncService:
    def __init__(self, client, custom_soap=None):
        self.method = ''
        self.client = client
        self.headers = {'Content-Type': 'text/xml; charset=UTF-8'}
        self.url = self.client.wsdl.location
        self.handle_custom(custom_soap)

    def handle_custom(self, soap):
        # handle custom urls and headers e.g SOAPAction
        return

    def beautify(self, _dict):
        # some beautyfying if you want
        env_key = next(key for key in _dict.keys() if 'env' in str(key).lower())
        body_key = next(key for key in _dict[env_key].keys() if 'body' in str(key).lower())
        sub_dict = _dict[env_key][body_key]
        while next(self.method.lower() in skey.lower() for skey in sub_dict.keys()):
            sub_dict_key = next(key for key in sub_dict.keys() if self.method.lower() in key.lower())
            sub_dict = sub_dict[sub_dict_key]
        if 'return' in sub_dict and hasattr(sub_dict['return'], 'keys'):
            return sub_dict['return']
        return sub_dict

    async def async_request(self, *args, **kwargs):
        msg = self.client.create_message(
            self.client.service,
            self.method,
            *args,
            **kwargs
        )
        encoded_request = etree.tostring(msg, encoding='utf8', method='xml')
        async with aiohttp.ClientSession() as session:
            async with session.post(self.url, headers=self.headers, data=encoded_request) as resp:
                result = await resp.text()
                _dict = xmltodict.parse(result)
                try:
                    _dict = self.beautify(_dict)
                except Exception as e:
                    # add loggers if you want
                    pass
                return _dict

    def __getattr__(self, item):
        self.method = item
        return self.async_request

Usage example:

async def f():
    transport = Transport(cache=SqliteCache())
    client = Client(prod_wsdl, transport=transport)

    client.aservice = common.ZeepAsyncService(client)
    result = await client.aservice.SomeMethod(someArgs)

In my case I'm using zeep client in sanic server. Result will always be dictionary.

[alex-kowalczyk]

The current (3.4.0) exception raising implemented in fa57c28e494f6730ebf0085aae533dca58a83392 "for reducing confusion" makes impossible to run in async code at all, even if the caller wraps the zeep constructor call into sync_to_async wrapper (available e.g. in asgiref package). I'd suggest making this exception a warning only.

[mvantellingen]

The next version (4.0) drops support for aiohttp altogether. One of the reasons development of zeep stagnated was the addition of all the async transports which made it a maintenance hell.

From zeep 4.0 and onwards only httpx will be supported for async operations (wsdl loading will be kept sync)

[decaz]

@mvantellingen was this function added to support async loading of WSDL? :)
https://github.com/mvantellingen/python-zeep/blob/6c9019cdd5052899c08f31a68bb3c1d39bfdc027/src/zeep/loader.py#L83

[mvantellingen]

Ah Yes. That should be removed. It was part of an experiment which reached a dead end. Mostly due to the fact that during the xsd parsing external content can be loaded. It requires a complete rewrite of the xsd parser

[decaz]

Maybe it worth to note the workaround in docs (here: https://docs.python-zeep.org/en/master/client.html?highlight=async#the-asyncclient) that if WSDL needs to be loaded asynchronously then you firstly should get it with any async http(s) client and pass received data to zeep.AsyncClient (instead of URL) at the moment of its initialization.

[mvantellingen]

That shouldn’t be needed anymore now right? (In 4.0)

[decaz]

Sorry, I didn't understand your question. What is "That"?

[mvantellingen]

Are you using the last version of zeep? 4.0?

[decaz]

Yep, 4.0.0.