Class for creating and managing recurring tasks. Attributes: func: The function to be executed repeatedly. delay: The time delay (in seconds) between function calls. task: The underlying task object.
| 16 | |
| 17 | |
| 18 | class RecurringTask: |
| 19 | """Class for creating and managing recurring tasks. |
| 20 | |
| 21 | Attributes: |
| 22 | func: The function to be executed repeatedly. |
| 23 | delay: The time delay (in seconds) between function calls. |
| 24 | task: The underlying task object. |
| 25 | """ |
| 26 | |
| 27 | def __init__(self, func: Callable, delay: timedelta) -> None: |
| 28 | logger.debug( |
| 29 | 'Calling RecurringTask.__init__(func={%s}, delay={%s})...', |
| 30 | func.__name__ if hasattr(func, '__name__') else func.__class__.__name__, |
| 31 | delay, |
| 32 | ) |
| 33 | self.func = func |
| 34 | self.delay = delay |
| 35 | self.task: asyncio.Task | None = None |
| 36 | |
| 37 | async def __aenter__(self) -> Self: |
| 38 | self.start() |
| 39 | return self |
| 40 | |
| 41 | async def __aexit__( |
| 42 | self, |
| 43 | exc_type: type[BaseException] | None, |
| 44 | exc_value: BaseException | None, |
| 45 | exc_traceback: TracebackType | None, |
| 46 | ) -> None: |
| 47 | await self.stop() |
| 48 | |
| 49 | async def _wrapper(self) -> None: |
| 50 | """Continuously execute the provided function with the specified delay. |
| 51 | |
| 52 | Run the function in a loop, waiting for the configured delay between executions. |
| 53 | Supports both synchronous and asynchronous functions. |
| 54 | """ |
| 55 | sleep_time_secs = self.delay.total_seconds() |
| 56 | while True: |
| 57 | await self.func() if inspect.iscoroutinefunction(self.func) else self.func() |
| 58 | await asyncio.sleep(sleep_time_secs) |
| 59 | |
| 60 | def start(self) -> None: |
| 61 | """Start the recurring task execution.""" |
| 62 | name = self.func.__name__ if hasattr(self.func, '__name__') else self.func.__class__.__name__ |
| 63 | self.task = asyncio.create_task( |
| 64 | self._wrapper(), |
| 65 | name=f'Task-recurring-{name}', |
| 66 | ) |
| 67 | |
| 68 | async def stop(self) -> None: |
| 69 | """Stop the recurring task execution.""" |
| 70 | if self.task: |
| 71 | self.task.cancel() |
| 72 | # Ensure the task has a chance to properly handle the cancellation and any potential exceptions. |
| 73 | await asyncio.gather(self.task, return_exceptions=True) |
no outgoing calls