asyncio_examples.py

#
# Working through some examples from python 3 asyncio docs
#
# https://docs.python.org/3/library/asyncio.html#module-asyncio
#

import asyncio
import random
import time


async def main_1():
    print("Hello ...")
    await asyncio.sleep(1)
    print("... World!")


async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)


async def main_2():
    print(f"started at {time.strftime('%X')}")
    await say_after(1, "hello")
    await say_after(2, "world")
    print(f"finished at {time.strftime('%X')}")


async def main_3():
    task1 = asyncio.create_task(say_after(1, "hello"))
    task2 = asyncio.create_task(say_after(2, "world"))
    print(f"started at {time.strftime('%X')}")
    await task1
    await task2
    print(f"finished at {time.strftime('%X')}")


def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    nap = random.randint(1, 10)
    print(f"sleeping for {nap}")
    time.sleep(nap)
    print(f"blocking_io complete at {time.strftime('%X')}")


async def main():
    task1 = asyncio.create_task(say_after(1, "hello"))
    task2 = asyncio.create_task(say_after(2, "world"))
    blocking_thread = asyncio.to_thread(blocking_io)
    print(f"started main at {time.strftime('%X')}")
    await asyncio.gather(task1, task2, blocking_thread)
    print(f"finished main at {time.strftime('%X')}")

asyncio.run(main())