A highlevel look at the implementation

import asyncio

import async_app.state as app_state
from async_app.app_factory import async_app_options

The following code is an excerpt of the real implementation to visualize the interface the user will deal with.

class AsyncApp(object):
    def __init__(self, **kwargs):
        self.task_descriptions = {
            "init": [],
            "continuous": [],
            "periodic": [],
            "cleanup": [],
        }

    def add_task_description(self, task_description):
        """Depending on the tasks kind, add it to its matching list."""
        ...

    async def run(self):
        """Run init tasks first, and after completion execute regular and periodic tasks."""
        ...
        
    async def task_monitor(self):
        """Offer a callback for a periodical to watch task execution."""
        ...

    def periodicals_monitor(self):
        """Offer a callback for periodicals monitoring that can be called periodically."""
        ...

    def exit(self, *args):
        """Exit the app cleanly."""
        app_state.keep_running = False
        

And providing some functions to be used with the AsyncApp …

def init_things(): ...
def run_forever(): ...
def ping(): ...
def cleanup(): ...

… it can be used like this:

app = AsyncApp()

app.add_task_description(
    {
        "kind": "init",
        "function": init_things,
    }
)

app.add_task_description(
    {
        "kind": "continuous",
        "function": run_forever,
    }
)

app.add_task_description(
    {
        "kind": "periodic",
        "function": ping,
        "call_every": 2,
    }
)

app.add_task_description(
    {
        "kind": "cleanup",
        "function": cleanup,
    }
)

A propper way to execute would now be calling asyncio.run(app.run()) and follow along what the app is doing. In this context awaiting the results is enough.

# asyncio.run(app.run())
await app.run()