Source code for galaxy.util.task

import logging
from threading import (

from galaxy.util import ExecutionTimer

log = logging.getLogger(__name__)

[docs]class IntervalTask:
[docs] def __init__(self, func, name="Periodic task", interval=3600, immediate_start=False, time_execution=False): """ Run an arbitrary function `func` every `interval` seconds. Set `immediate_start` to True to run `func` when task is started. """ self.func = func = name self.interval = interval self.time_execution = time_execution self.immediate_start = immediate_start self.event = Event() self.thread = Thread(,, daemon=True) self.running = False
[docs] def start(self): self.running = True self.thread.start()
def _exec(self): if self.time_execution: timer = ExecutionTimer() self.func() if self.time_execution: log.debug(f"Executed periodic task {} {timer}")
[docs] def run(self): if self.immediate_start: self._exec() while not self.event.is_set(): self.event.wait(self.interval) if self.running: self._exec()
[docs] def shutdown(self): self.running = False self.event.set() self.thread.join(5)