https://prefect.io logo
Title
v

Volker L

02/17/2023, 3:35 PM
I tried to run a simple local flow and it failed with this error.
RuntimeError: There is no current event loop in thread 'AnyIO worker thread'.
Inside one of my tasks, I am using asynio and aiohttp to request some data from a RestAPI.
1
n

Nate

02/17/2023, 3:36 PM
hi @Volker L - can you share your flow and how you tried to run it?
v

Volker L

02/17/2023, 3:39 PM
Sure. But I am using two of my one libraries, which makes it a bit complicated. I can try to set up a simple example flow, which might also fail 🙂
n

Nate

02/17/2023, 3:41 PM
okay! feel free to share what you can and I'll do my best to help
1
v

Volker L

02/17/2023, 3:52 PM
Ok. Writing a simple example, I was able to figured out what the problem is/was. 🙂 Here is the example:
from parallel_requests import parallel_requests, parallel_requests_async
from prefect import task, flow


@task
def download(urls):
    return parallel_requests(urls)

@task
async def download_async(urls):
    return await parallel_requests_async(urls)

@flow
def run():
    
    urls = ["<https://query2.finance.yahoo.com/v7/finance/quote?symbols=AAPL>"]*10
    res = download_async(urls)
    print(res[0])
    
    
if __name__=='__main__':
    run()
The library
parallel_requests
uses asyncio to run async functions. That seems to be not compatible with
AnyIO
(which prefect is using??). When I switch to
parallel_requests_async
, the flow is running without any problems. Note:
parallel_requests
is my own lib, which you can find here: https://github.com/legout/parallel-requests
n

Nate

02/17/2023, 3:59 PM
glad you figured it out!
AnyIO
(which prefect is using??).
yep! if you're curious how anyio is being used, you can check out this and/or this
🙏 1