FM / schedulers

/opt/imh-python/lib/python3.9/site-packages/apscheduler/schedulers/

/opt/imh-python/lib/python3.9/site-packages/apscheduler/schedulers

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
asyncio.py	1893	0644	edit dl rm
background.py	1476	0644	edit dl rm
base.py	48342	0644	edit dl rm
blocking.py	872	0644	edit dl rm
gevent.py	987	0644	edit dl rm
qt.py	1241	0644	edit dl rm
tornado.py	1890	0644	edit dl rm
twisted.py	1778	0644	edit dl rm
__init__.py	406	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/apscheduler/schedulers/background.py (1476B)

from threading import Event, Thread

from apscheduler.schedulers.base import BaseScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.util import asbool

class BackgroundScheduler(BlockingScheduler):
    """
    A scheduler that runs in the background using a separate thread
    (:meth:`~apscheduler.schedulers.base.BaseScheduler.start` will return immediately).

Extra options:

========== =============================================================================
    ``daemon`` Set the ``daemon`` option in the background thread (defaults to ``True``, see
               `the documentation
               <https://docs.python.org/3.4/library/threading.html#thread-objects>`_
               for further details)
    ========== =============================================================================
    """

_thread = None

def _configure(self, config):
        self._daemon = asbool(config.pop("daemon", True))
        super()._configure(config)

def start(self, *args, **kwargs):
        if self._event is None or self._event.is_set():
            self._event = Event()

BaseScheduler.start(self, *args, **kwargs)
        self._thread = Thread(
            target=self._main_loop, name="APScheduler", daemon=self._daemon
        )
        self._thread.start()

def shutdown(self, *args, **kwargs):
        super().shutdown(*args, **kwargs)
        self._thread.join()
        del self._thread

from threading import Event, Thread from apscheduler.schedulers.base import BaseScheduler from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.util import asbool class BackgroundScheduler(BlockingScheduler): """ A scheduler that runs in the background using a separate thread (:meth:`~apscheduler.schedulers.base.BaseScheduler.start` will return immediately). Extra options: ========== ============================================================================= ``daemon`` Set the ``daemon`` option in the background thread (defaults to ``True``, see `the documentation `_ for further details) ========== ============================================================================= """ _thread = None def _configure(self, config): self._daemon = asbool(config.pop("daemon", True)) super()._configure(config) def start(self, *args, **kwargs): if self._event is None or self._event.is_set(): self._event = Event() BaseScheduler.start(self, *args, **kwargs) self._thread = Thread( target=self._main_loop, name="APScheduler", daemon=self._daemon ) self._thread.start() def shutdown(self, *args, **kwargs): super().shutdown(*args, **kwargs) self._thread.join() del self._thread