Python asyncio event loop in a separate thread

asyncio_loop_in_thread.py

"""
This gist shows how to run asyncio loop in a separate thread.
It could be useful if you want to mix sync and async code together.
Python 3.7+
"""
import asyncio
from datetime import datetime
from threading import Thread
from typing import Tuple, List, Iterable

import httpx

URLS = [
    "https://pypi.org",
    "https://python.org",
    "https://google.com",
    "https://amazon.com",
    "https://reddit.com",
    "https://stackoverflow.com",
    "https://ubuntu.com",
    "https://github.com",
    "https://microsoft.com",
]


def start_background_loop(loop: asyncio.AbstractEventLoop) -> None:
    asyncio.set_event_loop(loop)
    loop.run_forever()


async def fetch(url: str) -> Tuple[str, int]:
    """Does HTTP get on url and returns url and status code"""
    async with httpx.AsyncClient() as session:
        response = await session.get(url)
        return url, response.status_code


async def fetch_all_urls(urls: Iterable[str]) -> List[Tuple[str, int]]:
    """Fetch all urls from the list of urls

    It is done concurrently and combined into a single coroutine"""
    tasks = [asyncio.create_task(fetch(url)) for url in urls]
    results = await asyncio.gather(*tasks)
    return results


def main() -> None:
    loop = asyncio.new_event_loop()
    t = Thread(target=start_background_loop, args=(loop,), daemon=True)
    t.start()

    start_time = datetime.now()

    task = asyncio.run_coroutine_threadsafe(fetch_all_urls(URLS), loop)
    for url, status_code in task.result():
        print(f"{url} -> {status_code}")

    exec_time = (datetime.now() - start_time).total_seconds()
    print(f"It took {exec_time:,.2f} seconds to run")
    loop.stop()


if __name__ == "__main__":
    main()

Totally agree with this – it just isn’t clear for newcomers. From: James Pop ***@***.***> Date: Tuesday, 31 January 2023 at 04:51 To: jryebread ***@***.***> Cc: Comment ***@***.***> Subject: Re: dmfigol/asyncio_loop_in_thread.py @jryebread commented on this gist.

…

________________________________ Thank you, simple and straight to the point, the docs online for async stuff are trash — Reply to this email directly, view it on GitHub<https://gist.github.com/3e7d5b84a16d076df02baa9f53271058#gistcomment-4454151> or unsubscribe<https://github.com/notifications/unsubscribe-auth/AOX6R36H7QOCJMBTJFA2AB3WVCK3LBFKMF2HI4TJMJ2XIZLTSKBKK5TBNR2WLJDHNFZXJJDOMFWWLK3UNBZGKYLEL52HS4DFQKSXMYLMOVS2I5DSOVS2I3TBNVS3W5DIOJSWCZC7OBQXE5DJMNUXAYLOORPWCY3UNF3GS5DZVRZXKYTKMVRXIX3UPFYGLK2HNFZXIQ3PNVWWK3TUUZ2G64DJMNZZDAVEOR4XAZNEM5UXG5FFOZQWY5LFVA4TENRWGA4TENVHORZGSZ3HMVZKMY3SMVQXIZI>. You are receiving this email because you commented on the thread. Triage notifications on the go with GitHub Mobile for iOS<https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android<https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>.

Nice. all asyncio tutorials talk about run threads in a syncio program. but your example is so much nice. I always want to create a thread for asyncio and send all asynchronous task in to it and works with synchronous in the main thread. but I didn't know how to do it exactly. because I want to build a gui app that need some asynchronous results. and all gui framework have their own event loop. Thanks a lot.

so do i.

Thanks so much! I was trying to get this to work for 3 days, and when I finally found this, it worked first try!

I wrote some helper functions for this:

from typing import Coroutine
import asyncio
from asyncio import Future, AbstractEventLoop
from threading import Thread


def create_event_loop_thread() -> AbstractEventLoop:
    """
    From https://gist.github.com/dmfigol/3e7d5b84a16d076df02baa9f53271058
    """
    def start_background_loop(loop: AbstractEventLoop) -> None:
        asyncio.set_event_loop(loop)
        loop.run_forever()

    eventloop = asyncio.new_event_loop()
    thread = Thread(target=start_background_loop, args=(eventloop,), daemon=True)
    thread.start()
    return eventloop

def run_coroutine_in_thread(coro: Coroutine) -> Future:
    """
    From https://gist.github.com/dmfigol/3e7d5b84a16d076df02baa9f53271058
    """
    return asyncio.run_coroutine_threadsafe(coro, self.eventloop)

Thank you so much! It really works on the first try. I have been looking for this solution all this time, now my bug is solved. Thank you!!

Saved my life 😯