Skip to the content.

Asynchronous I/O - Non-blocking main thread

Execute IO-bound functions/methods that would otherwise block the event loop if they were ran in the main thread.

Solution

"""Asynchronous I/O - Non-blocking main thread.
"""

import asyncio
import logging
import time

logging.basicConfig(
    level=logging.DEBUG, style='{', format='[{threadName} ({thread})] {message}'
)


def blocking_io():
    logging.debug(f'start blocking_io at {time.strftime("%X")}')
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    logging.debug(f'blocking_io complete at {time.strftime("%X")}')


async def main():
    logging.debug(f'started main at {time.strftime("%X")}')
    await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1),  # type: ignore
    )
    logging.debug(f'finished main at {time.strftime("%X")}')


asyncio.run(main())  # Python 3.7+

See source code

References