How to call a async function from a synchronized code Python

Question

So I'm locked to a python 3.6.2 interpreter that follows my desktop application.

What I want is to call an async function from a synchronized method or function.

When calling the python function from the desktop application it has to be a normal function which can not be awaited.

From the desktop application I am able to send a list of urls, and what I want is to send back response from every url in an async matter.

here is my try I've marked the SyntaxError which I don't know how to bypass.

import fmeobjects
import asyncio
import aiohttp
import async_timeout
logger = fmeobjects.FMELogFile()
timeout = 10

class FeatureProcessor(object):
    def __init__(self):
        pass
    def input(self, feature):
        urls_and_coords = zip(feature.getAttribute('_list{}._wms'),\
        feature.getAttribute('_list{}._xmin'),\
        feature.getAttribute('_list{}._ymin'),\
        feature.getAttribute('_list{}._xmax'),\
        feature.getAttribute('_list{}._ymax'))
        -> SyntaxError: newfeature = await main(urls_and_coords)
        self.pyoutput(newfeature)
        
    def close(self):
       pass 

async def main(urls):
    loop = asyncio.get_event_loop()
    async with aiohttp.ClientSession(loop=loop) as session:
        feature = loop.run_until_complete(fetch_all(session, urls, loop))
        return feature
        
async def fetch_all(session, urls, loop):
    results = await asyncio.gather(*[loop.create_task(fetch(session, url)) for url in urls])
    return results
    

async def fetch(session, url):
    with async_timeout.timeout(10):
        async with session.get(url[0]) as response:
            newFeature = fmeobjects.FMEFeature()
            response_data = await response
            newFeature.setAttribute('response', response_data)
            newFeature.setAttribute('_xmin',url[1])
            newFeature.setAttribute('_xmax',url[2])
            newFeature.setAttribute('_ymin',url[3])
            newFeature.setAttribute('_ymax',url[4])
            return newFeature

I have tried making these changes:

import fme
import fmeobjects
import asyncio
import aiohttp
import async_timeout
logger = fmeobjects.FMELogFile()

class FeatureProcessor(object):
    def __init__(self):
        pass
    def input(self, feature):
        urls_and_coords = zip(feature.getAttribute('_list{}._wms'),\
        feature.getAttribute('_list{}._xmin'),\
        feature.getAttribute('_list{}._ymin'),\
        feature.getAttribute('_list{}._xmax'),\
        feature.getAttribute('_list{}._ymax'))
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(main(loop, urls_and_coords))
        #feature.setAttribute('result',result)
        self.pyoutput(feature)
        
    def close(self):
       pass 

async def main(loop, urls):
    async with aiohttp.ClientSession(loop=loop) as session:
        return await fetch_all(session, urls, loop)

        
async def fetch_all(session, urls, loop):
    results = await asyncio.gather(*[loop.create_task(fetch(session, url)) for url in urls])
    return results
    

async def fetch(session, url):
    with async_timeout.timeout(10):
        async with session.get(url[0]) as response:
            #newFeature = fmeobjects.FMEFeature()
            response = await response
            #newFeature.setAttribute('response', response_data)
            #newFeature.setAttribute('_xmin',url[1])
            #newFeature.setAttribute('_xmax',url[2])
            #newFeature.setAttribute('_ymin',url[3])
            #newFeature.setAttribute('_ymax',url[4])
            return response, url[1], url[2], url[3], url[4]

but now I end up with this error:

Python Exception <TypeError>: object ClientResponse can't be used in 'await' 
expression
Traceback (most recent call last):
  File "<string>", line 20, in input
  File "asyncio\base_events.py", line 467, in run_until_complete
  File "<string>", line 29, in main
  File "<string>", line 33, in fetch_all
  File "<string>", line 41, in fetch
TypeError: object ClientResponse can't be used in 'await' expression

You may want to have a look at the ``trio`` library. It has a much more straightforward interface than the ``asyncio`` standard library. — MisterMiyagi, Aug 09 '18 at 08:55
Cool, looks like it implements run similar to asyncio in python 3.7. I will have a look at this. — Paal Pedersen, Aug 09 '18 at 09:37

score 37 · Answer 1 · edited Dec 09 '19 at 23:41

37

@deceze answer is probably the best you can do in Python 3.6. But in Python 3.7, you could directly use asyncio.run in the following way:

newfeature = asyncio.run(main(urls))

It will properly create, handle, and close an event_loop.

edited Dec 09 '19 at 23:41

Neuron - Freedom for Ukraine

4,303
4
29
51

answered Aug 09 '18 at 08:52

Francis Colas

2,991
2
24
31

1

But what if the code is already running in an `asyncio.run` call? If you use `asyncio.run` inside a function that is invoked with `asyncio.run` then you get `RuntimeError: asyncio.run() cannot be called from a running event loop` – birgersp Oct 31 '21 at 12:11
2

@birgersp If you are already inside an event_loop you can simply call it via `result = await main(urls)`. – mgutsche Nov 08 '21 at 15:57

score 23 · Answer 2 · answered Aug 09 '18 at 08:34

23

You would use an event loop to execute the asynchronous function to completion:

newfeature = asyncio.get_event_loop().run_until_complete(main(urls_and_coords))

(This technique is already used inside main. And I'm not sure why, since main is async you could/should use await fetch_all(...) there.)

answered Aug 09 '18 at 08:34

deceze

491,798
79
706
853

2

But then I probably need to rewrite main, since it already has an event_loop? – Paal Pedersen Aug 09 '18 at 08:38
Interesting point, I'm not actually sure off the top of my head whether that would cause any issues. But as I wrote, it makes little sense to use `run_until_complete` inside an `async` function to begin with, you should simply `await` it. – deceze Aug 09 '18 at 08:41
It works for me. Note that in the case you have not event_loop and so you have the error : "RuntimeError: There is no current event loop in thread 'Thread-n'", you can add `asyncio.set_event_loop(asyncio.new_event_loop())` in your function to set an event loop. – 1ronmat Apr 17 '20 at 09:17

score 9 · Answer 3 · answered May 03 '20 at 04:38

9

Another option that could be useful is the syncer PyPI package:

from syncer import sync

@sync
async def print_data():
    print(await get_data())

print_data()  # Can be called synchronously

answered May 03 '20 at 04:38

Matthew D. Scholefield

2,610
2
29
39

How come this answer has so little upvotes? Thanks, it's great for writing packages with `aiohttp` and for beginners taking their first steps with asynchronous features in Python. – Konrad Mar 30 '22 at 11:19

How to call a async function from a synchronized code Python

3 Answers3

Linked

Related