the script demonstrates how to implement the async mechanisms in Python

async.py

#!/bin/python

# description: 
#   |- the script demonstrates how to implement the async mechanisms in Python
#   |- asyncio was implemented <= python 3.6 and >= python 3.7
# maintainer: JianKai Wang <[email protected]>

import sys
import time
import asyncio

def GetVersionForAsync():
  """Get the current python version for async.

  Args:
    None

  Returns:
    flag: lower (py<=3.6), higher (py>=3.7), others
    verInfo: the major_version.minor_version string
  """
  major, minor, _, _, _ = sys.version_info
  verInfo = "{}.{}".format(major, minor)

  if int(major) == 3:
    if int(minor) >= 7:
      return "higher", verInfo
    else:
      return "lower", verInfo
  else:
    return "others", verInfo

async def callee(sleepTime=3.0):
  """the example of callee
  
  Args:
    sleepTime: the parameter for simulating the time cost of functions
  
  Returns:
    state: True for success, otherwise False
    timeCost: the time cost
  """
  state = False
  __startTime = time.perf_counter()
  try:
    await asyncio.sleep(sleepTime)
    state = True
  except Exception as err:
    print("Failed in callee with error {}.".format(err))
    state = False
  finally:
    timeCost = time.perf_counter() - __startTime
    return state, timeCost


async def HigherCaller(sleepTime=3.0, timeOut=5.0):
  """the example of caller >= Python 3.7
  
  Args:
    sleepTime: the parameter for simulating the time cost of functions
    timeOut: the timeout in second

  Returns:
    results: the simple result
  """
  __startTime = time.perf_counter()
  try:
    print("Try to keep {} sleep second and to limit in {} second.".format(
      sleepTime, timeOut))
    task = asyncio.create_task(callee(sleepTime=sleepTime))
    await asyncio.wait_for(task, timeout=timeOut)
  except asyncio.TimeoutError:
    print("Task in HigherCaller was timeout.")
    print("Time Cost: {}".format(time.perf_counter() - __startTime))
    return False

  if task.done():
    state, timeCost = task.result()
    print("Time Cost: {}".format(timeCost))
    return state
  else:
    print("Task in HigherCaller was not successful.")
    return False

async def LowerCaller(loop, sleepTime=3.0, timeOut=5.0):
  """the example of caller <= Python 3.6
  
  Args:
    sleepTime: the parameter for simulating the time cost of functions
    timeOut: the timeout in second

  Returns:
    results: the simple result  
  """
  __startTime = time.perf_counter()
  try:
    print("Try to keep {} sleep second and to limit in {} second.".format(
      sleepTime, timeOut))
    task = loop.create_task(callee(sleepTime=sleepTime))
    await asyncio.wait_for(task, timeout=timeOut)
  except asyncio.TimeoutError:
    print("Task in LowerCaller was timeout.")
    print("Time Cost: {}".format(time.perf_counter() - __startTime))
    return False
    
  if task.done():
    state, timeCost = task.result()
    print("Time Cost: {}".format(timeCost))
    return state
  else:
    print("Task in LowerCaller was not successful.")
    return False  

def main():
  """the entry for demonstrating async"""

  flag, verInfo = GetVersionForAsync()
  print("The Python Version: {}".format(verInfo))

  if flag == "higher" :
    import nest_asyncio
    nest_asyncio.apply()

    state = asyncio.run(HigherCaller(sleepTime=10.0, timeOut=5.0))
    print("State in HigherCaller: {}".format(state))
  elif flag == "lower":
    loop = asyncio.get_event_loop()
    state = loop.run_until_complete(\
      LowerCaller(loop=loop, sleepTime=3.0, timeOut=5.0))
    print("State in LowerCaller: {}".format(state))    
  else:
    print("Unsupported version.")

  sys.exit(0)

if __name__ == "__main__":
  main()