Python3 Asyncio runner with grafecfull exit

async_runner.py

import signal
import asyncio

from prometheus_client import start_http_server as start_prometheus_server


EXIT_EVENT = asyncio.Event()


async def infinite_run():
    while not EXIT_EVENT.is_set():
        # do stuff...
        await asyncio.sleep(1)
    else:
        print("Stopped gracefully")


async def run():
    loop = asyncio.get_running_loop()
    signals = (signal.SIGHUP, signal.SIGTERM, signal.SIGINT)
    for s in signals:
        loop.add_signal_handler(s, lambda: EXIT_EVENT.set())


    # Run in prometheus in separate thread to avoid blocking
    loop.run_in_executor(None, start_prometheus_server, 80)

    # Run tasks
    tasks = []
    tasks.append(loop.create_task(infinite_run()))
    # more tasks...


    # Wait for exit event
    await EXIT_EVENT.wait()
    logger.info('Shutdown signal received, stopping tasks gracefully...')
    # Ensure all tasks complete
    await asyncio.gather(*tasks, return_exceptions=True)
    logger.info('All tasks stopped cleanly.')

if __name__ == '__main__':
    asyncio.run(run())