Python3 Asyncio Main Template

asyncio_main_example.py

# GIT Home: https://gist.github.com/JavaScriptDude/f3c837fdabaf0e0a72e39ff319c7a3f0
# NOTE: This version works with Python 3.7.9.
#       For Python 3.11, see 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)
        
        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)

    # Call main
    try :
        asyncio.get_event_loop().run_until_complete(main())

    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.Task:
            for task in asyncio.Task.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:])

This template will ensure that program bootstrapping like args parsing, and main error handling is done outside of asyncio, and makes the output of the program much cleaner.

Its important to not call sys.exit() from within asyncio processes as any existing tasks and dependent processes may trigger dumping of extra log messages to stdout or stderr after the exit is triggered. By using raise ExitProgram(), from main() you can still fail fast with clean messaging and allow clean shutdown of processes before final messages are sent out.