Asyncio Example for Python 3.11+

asyncio_main_example_3_11.py

# GIT Home: https://gist.github.com/JavaScriptDude/f673980de8e27a39cbffff55dd0c63b2

import sys
import traceback
import asyncio
_verbose = ("--verbose" in sys.argv)
opts = None

# Main - For all your business logic
# Note: do not call sys.exit() directly from here, instead use raise ExitProgram(msg=<msg>, code=<code>)
# This ensures that cleanup code executed in finally block runs before sys.exit() is called
async def main():
    if _verbose: print(f"main() called. opts: {opts}")
    try:

        print("... add code here ...")

        if False: # Example of early program exit
            raise ExitProgram(msg="FOOBAR", code=2)
        
        await asyncio.sleep(10)
        
        print("... add more code here ...")

    finally:
        # Trigger any explicity process closures here.
        # Eg. for Pyppeteer, run browser.close()
        pass


# Initialization and Startup
def bootstrap(argv):
    global opts
    if _verbose: print(f"bootstrap() called. argv: {argv}")
    # process cli args here
    try:
        opts = {"opt1": "val1", "verbose": _verbose}

    except Exception as e:
        print(f"Fatal exception during opts processing(): {_get_last_exc()}")
        _exit_program(1)


    # Setup our event loop
    loop = asyncio.get_event_loop()

    # Start main
    asyncio.ensure_future(main())

    try :
        loop.run_forever()

    except ExitProgram as ep: # Handle Exit 
        print(ep.message)
        _exit_program(ep.code)
        
    except Exception as e:
        print(f"Fatal exception during main(): {_get_last_exc()}")
        _exit_program(1)

    except KeyboardInterrupt:
        print("Received exit, exiting")
        _exit_program(0)

    except:
        print(f"Unexpected Error occurred from within program: {_get_last_exc()}")
        _exit_program(1)

    if _verbose: print("Script exiting cleanly")
    
    _exit_program(0)



# Utilities
def _get_last_exc():
    exc_type, exc_value, exc_traceback = sys.exc_info()
    sTB = '\n'.join(traceback.format_tb(exc_traceback))
    return f"{exc_type}\n - msg: {exc_value}\n stack: {sTB}"


def _exit_program(code=1):
    if _verbose: 
        print(f"_exit_program() called code = {code}")

    _stdout_bk = _stderr_bk = None
    try:
        # Trap any stderr / out's from tasks
        _stdout_bk = sys.stdout
        _stderr_bk = sys.stderr
        
        sys.stdout=DEVNULL
        sys.stderr=DEVNULL

        # kill all active asyncio Tasks
        if asyncio.get_event_loop().is_running():
            for task in asyncio.all_tasks():
                try:
                    task.cancel()
                except Exception as ex:
                    pass

    finally:
        sys.stdout = _stdout_bk
        sys.stderr = _stderr_bk

    # Shut down
    if _verbose: print(f"exiting with code {code}")

    # flush stderr and stdout
    sys.stdout.flush()
    sys.stderr.flush()

    sys.exit(code)


class DevNull():
    def __init__(self, **args): pass
    def write(self, s): return 0
    def writelines(self, lines): pass

DEVNULL = DevNull()



class ExitProgram(Exception):
    message: str = None
    code: int = 1
    def __init__(self, msg: str, code: int = 1):
        super().__init__(msg)
        assert isinstance(msg, str) and not msg.strip() == '' \
                ,"msg must be a non blank string"
        assert isinstance(code, int) \
                ,"code must be an integer"
        self.message = msg
        self.code = code


if __name__ == '__main__':
    bootstrap(sys.argv[1:])