malepati/custom_template_working

1import asyncio2from asyncio import Task3from typing import Any, Callable, Coroutine, Optional4 5 6class ConcurrentService:7    def __init__(self):8        self._background_tasks = set[Task]()9 10    def run_task(11        self,12        delay: Optional[int],13        callable: Callable[..., Coroutine[Any, Any, Any]],14        *args,15        **kwargs,16    ):17        async def wrapper():18            if delay:19                await asyncio.sleep(delay)20            await callable(*args, **kwargs)21 22        task = asyncio.create_task(wrapper())23 24        print(f"Running task: {task} - executing {callable.__name__}")25 26        self._background_tasks.add(task)27        task.add_done_callback(self.on_task_done)28 29    def on_task_done(self, task: Task):30        print(f"Task done: {task}")31 32        self._background_tasks.discard(task)33 34 35CONCURRENT_SERVICE = ConcurrentService()36