A Python script showcasing how to run two independent loops simultaneously using asyncio, with continuous task execution and error handling.

independent_async_loops.py

import logging
import asyncio
import time

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)


def time_check_decorator(func):
    def wrapper(*args, **kwargs):
        print(f'{func.__name__} started at {time.strftime("%X")}')
        func(*args, **kwargs)
        print(f'{func.__name__} ended at {time.strftime("%X")}')

    return wrapper


@time_check_decorator
def task1():
    count = 0
    for i in range(100000000):
        count += 1
    print(f'count: {count}')
    raise ValueError('Error in task1')


@time_check_decorator
def task2():
    count = 0
    for i in range(30000000):
        count += 1
    print(f'count: {count}')


async def looper(func):
    loop = asyncio.get_running_loop()
    while True:
        task = loop.run_in_executor(None, func)
        try:
            await task
        except ValueError as e:
            logger.error(f'Error in task: {e}')
            return
        await asyncio.sleep(2)


async def main():
    task1_loop_ = asyncio.create_task(looper(task1))
    task2_loop_ = asyncio.create_task(looper(task2))

    await task1_loop_
    await task2_loop_


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