433 lines
16 KiB
Python
433 lines
16 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: UTF-8 -*-
|
|
|
|
"""
|
|
task (Task Module)
|
|
==================
|
|
|
|
**Author:**
|
|
|
|
* Dirk Alders <sudo-dirk@mount-mockery.de>
|
|
|
|
**Description:**
|
|
|
|
This Module supports helpfull classes for queues, tasks, ...
|
|
|
|
**Submodules:**
|
|
|
|
* :class:`task.crontab`
|
|
* :class:`task.delayed`
|
|
* :class:`task.periodic`
|
|
* :class:`task.queue`
|
|
* :class:`task.threaded_queue`
|
|
|
|
**Unittest:**
|
|
|
|
See also the :download:`unittest <task/_testresults_/unittest.pdf>` documentation.
|
|
|
|
**Module Documentation:**
|
|
|
|
"""
|
|
__DEPENDENCIES__ = []
|
|
|
|
import logging
|
|
import sys
|
|
import threading
|
|
import time
|
|
if sys.version_info >= (3, 0):
|
|
from queue import PriorityQueue
|
|
from queue import Empty
|
|
else:
|
|
from Queue import PriorityQueue
|
|
from Queue import Empty
|
|
|
|
try:
|
|
from config import APP_NAME as ROOT_LOGGER_NAME
|
|
except ImportError:
|
|
ROOT_LOGGER_NAME = 'root'
|
|
logger = logging.getLogger(ROOT_LOGGER_NAME).getChild(__name__)
|
|
|
|
__DESCRIPTION__ = """The Module {\\tt %s} is designed to help with task issues like periodic tasks, delayed tasks, queues, threaded queues and crontabs.
|
|
For more Information read the documentation.""" % __name__.replace('_', '\_')
|
|
"""The Module Description"""
|
|
__INTERPRETER__ = (2, 3)
|
|
"""The Tested Interpreter-Versions"""
|
|
|
|
|
|
class queue(object):
|
|
"""
|
|
Class to execute queued callbacks.
|
|
|
|
:param bool expire: The default value for expire. See also :py:func:`expire`.
|
|
|
|
**Example:**
|
|
|
|
.. literalinclude:: task/_examples_/tqueue.py
|
|
|
|
Will result to the following output:
|
|
|
|
.. literalinclude:: task/_examples_/tqueue.log
|
|
"""
|
|
class job(object):
|
|
def __init__(self, priority, callback, *args, **kwargs):
|
|
self.time = time.time()
|
|
self.priority = priority
|
|
self.callback = callback
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
|
|
def run(self, queue):
|
|
self.callback(queue, *self.args, **self.kwargs)
|
|
|
|
def __lt__(self, other):
|
|
if self.priority != other.priority:
|
|
return self.priority < other.priority
|
|
else:
|
|
return self.time < other.time
|
|
|
|
def __init__(self, expire=True):
|
|
self.__expire = expire
|
|
self.__stop = False
|
|
self.queue = PriorityQueue()
|
|
|
|
def clean_queue(self):
|
|
"""
|
|
This Methods removes all jobs from the queue.
|
|
|
|
.. note:: Be aware that already running jobs will not be terminated.
|
|
"""
|
|
while not self.queue.empty():
|
|
try:
|
|
self.queue.get(False)
|
|
except Empty: # This block is hard to reach for a testcase, but is
|
|
continue # needed, if the thread runs dry while cleaning the queue.
|
|
self.queue.task_done()
|
|
|
|
def enqueue(self, priority, callback, *args, **kwargs):
|
|
"""
|
|
This enqueues a given callback.
|
|
|
|
:param number priority: The priority indication number of this task. The lowest value will be queued first.
|
|
:param callback callback: Callback to be executed
|
|
:param args args: Arguments to be given to callback
|
|
:param kwargs kwargs: Keword Arguments to be given to callback
|
|
|
|
.. note:: Callback will get this instance as first argument, followed by :py:data:`args` und :py:data:`kwargs`.
|
|
"""
|
|
self.queue.put(self.job(priority, callback, *args, **kwargs))
|
|
|
|
def qsize(self):
|
|
return self.queue.qsize()
|
|
|
|
def run(self):
|
|
"""
|
|
This starts the execution of the queued callbacks.
|
|
"""
|
|
self.__stop = False
|
|
while not self.__stop:
|
|
try:
|
|
self.queue.get(timeout=0.1).run(self)
|
|
except Empty:
|
|
if self.__expire:
|
|
break
|
|
if type(self) is threaded_queue:
|
|
self.thread = None
|
|
|
|
def expire(self):
|
|
"""
|
|
This sets the expire flag. That means that the process will stop after queue gets empty.
|
|
"""
|
|
self.__expire = True
|
|
|
|
def stop(self):
|
|
"""
|
|
This sets the stop flag. That means that the process will stop after finishing the active task.
|
|
"""
|
|
self.__stop = True
|
|
|
|
|
|
class threaded_queue(queue):
|
|
"""Class to execute queued callbacks in a background thread (See also parent :py:class:`queue`).
|
|
|
|
:param bool expire: The default value for expire. See also :py:func:`queue.expire`.
|
|
|
|
**Example:**
|
|
|
|
.. literalinclude:: task/_examples_/threaded_queue.py
|
|
|
|
Will result to the following output:
|
|
|
|
.. literalinclude:: task/_examples_/threaded_queue.log
|
|
"""
|
|
def __init__(self, expire=False):
|
|
queue.__init__(self, expire=expire)
|
|
self.thread = None
|
|
|
|
def run(self):
|
|
if self.thread is None:
|
|
self.thread = threading.Thread(target=self._start, args=())
|
|
self.thread.daemon = True # Daemonize thread
|
|
self.thread.start() # Start the execution
|
|
|
|
def join(self):
|
|
"""
|
|
This blocks till the queue is empty.
|
|
|
|
.. note:: If the queue does not run dry, join will block till the end of the days.
|
|
"""
|
|
self.expire()
|
|
if self.thread is not None:
|
|
self.thread.join()
|
|
|
|
def stop(self):
|
|
queue.stop(self)
|
|
self.join()
|
|
|
|
def _start(self):
|
|
queue.run(self)
|
|
|
|
|
|
class periodic(object):
|
|
"""
|
|
Class to execute a callback cyclicly.
|
|
|
|
:param float cycle_time: Cycle time in seconds -- callback will be executed every *cycle_time* seconds
|
|
:param callback callback: Callback to be executed
|
|
:param args args: Arguments to be given to the callback
|
|
:param kwargs kwargs: Keword Arguments to be given to callback
|
|
|
|
.. note:: The Callback will get this instance as first argument, followed by :py:data:`args` und :py:data:`kwargs`.
|
|
|
|
**Example:**
|
|
|
|
.. literalinclude:: task/_examples_/periodic.py
|
|
|
|
Will result to the following output:
|
|
|
|
.. literalinclude:: task/_examples_/periodic.log
|
|
"""
|
|
def __init__(self, cycle_time, callback, *args, **kwargs):
|
|
self._lock = threading.Lock()
|
|
self._timer = None
|
|
self.callback = callback
|
|
self.cycle_time = cycle_time
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
self._stopped = True
|
|
self._last_tm = None
|
|
self.dt = None
|
|
|
|
def join(self):
|
|
"""
|
|
This blocks till the cyclic task is terminated.
|
|
|
|
.. note:: Using join means that somewhere has to be a condition calling :py:func:`stop` to terminate. Otherwise :func:`task.join` will never return.
|
|
"""
|
|
while not self._stopped:
|
|
time.sleep(.1)
|
|
|
|
def run(self):
|
|
"""
|
|
This starts the cyclic execution of the given callback.
|
|
"""
|
|
if self._stopped:
|
|
self._set_timer(force_now=True)
|
|
|
|
def stop(self):
|
|
"""
|
|
This stops the execution of any further task.
|
|
"""
|
|
self._lock.acquire()
|
|
self._stopped = True
|
|
if self._timer is not None:
|
|
self._timer.cancel()
|
|
self._lock.release()
|
|
|
|
def _set_timer(self, force_now=False):
|
|
"""
|
|
This sets the timer for the execution of the next task.
|
|
"""
|
|
self._lock.acquire()
|
|
self._stopped = False
|
|
if force_now:
|
|
self._timer = threading.Timer(0, self._start)
|
|
else:
|
|
self._timer = threading.Timer(self.cycle_time, self._start)
|
|
self._timer.start()
|
|
self._lock.release()
|
|
|
|
def _start(self):
|
|
tm = time.time()
|
|
if self._last_tm is not None:
|
|
self.dt = tm - self._last_tm
|
|
self._set_timer(force_now=False)
|
|
self.callback(self, *self.args, **self.kwargs)
|
|
self._last_tm = tm
|
|
|
|
|
|
class delayed(periodic):
|
|
"""Class to execute a callback a given time in the future. See also parent :py:class:`periodic`.
|
|
|
|
:param float time: Delay time for execution of the given callback
|
|
:param callback callback: Callback to be executed
|
|
:param args args: Arguments to be given to callback
|
|
:param kwargs kwargs: Keword Arguments to be given to callback
|
|
|
|
**Example:**
|
|
|
|
.. literalinclude:: task/_examples_/delayed.py
|
|
|
|
Will result to the following output:
|
|
|
|
.. literalinclude:: task/_examples_/delayed.log
|
|
"""
|
|
def run(self):
|
|
"""
|
|
This starts the timer for the delayed execution.
|
|
"""
|
|
self._set_timer(force_now=False)
|
|
|
|
def _start(self):
|
|
self.callback(*self.args, **self.kwargs)
|
|
self.stop()
|
|
|
|
|
|
class crontab(periodic):
|
|
"""Class to execute a callback at the specified time conditions. See also parent :py:class:`periodic`.
|
|
|
|
:param accuracy: Repeat time in seconds for background task checking event triggering. This time is the maximum delay between specified time condition and the execution.
|
|
:type accuracy: float
|
|
|
|
**Example:**
|
|
|
|
.. literalinclude:: task/_examples_/crontab.py
|
|
|
|
Will result to the following output:
|
|
|
|
.. literalinclude:: task/_examples_/crontab.log
|
|
"""
|
|
ANY = '*'
|
|
"""Constant for matching every condition."""
|
|
|
|
class cronjob(object):
|
|
"""Class to handle cronjob parameters and cronjob changes.
|
|
|
|
:param minute: Minute for execution. Either 0...59, [0...59, 0...59, ...] or :py:const:`crontab.ANY` for every Minute.
|
|
:type minute: int, list, str
|
|
:param hour: Hour for execution. Either 0...23, [0...23, 0...23, ...] or :py:const:`crontab.ANY` for every Hour.
|
|
:type hour: int, list, str
|
|
:param day_of_month: Day of Month for execution. Either 0...31, [0...31, 0...31, ...] or :py:const:`crontab.ANY` for every Day of Month.
|
|
:type day_of_month: int, list, str
|
|
:param month: Month for execution. Either 0...12, [0...12, 0...12, ...] or :py:const:`crontab.ANY` for every Month.
|
|
:type month: int, list, str
|
|
:param day_of_week: Day of Week for execution. Either 0...6, [0...6, 0...6, ...] or :py:const:`crontab.ANY` for every Day of Week.
|
|
:type day_of_week: int, list, str
|
|
:param callback: The callback to be executed. The instance of :py:class:`cronjob` will be given as the first, args and kwargs as the following parameters.
|
|
:type callback: func
|
|
|
|
.. note:: This class should not be used stand alone. An instance will be created by adding a cronjob by using :py:func:`crontab.add_cronjob()`.
|
|
"""
|
|
class all_match(set):
|
|
"""Universal set - match everything"""
|
|
def __contains__(self, item):
|
|
(item)
|
|
return True
|
|
|
|
def __init__(self, minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs):
|
|
self.set_trigger_conditions(minute or crontab.ANY, hour or crontab.ANY, day_of_month or crontab.ANY, month or crontab.ANY, day_of_week or crontab.ANY)
|
|
self.callback = callback
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
self.__last_cron_check_time__ = None
|
|
self.__last_execution__ = None
|
|
|
|
def set_trigger_conditions(self, minute=None, hour=None, day_of_month=None, month=None, day_of_week=None):
|
|
"""This Method changes the execution parameters.
|
|
|
|
:param minute: Minute for execution. Either 0...59, [0...59, 0...59, ...] or :py:const:`crontab.ANY` for every Minute.
|
|
:type minute: int, list, str
|
|
:param hour: Hour for execution. Either 0...23, [0...23, 0...23, ...] or :py:const:`crontab.ANY` for every Hour.
|
|
:type hour: int, list, str
|
|
:param day_of_month: Day of Month for execution. Either 0...31, [0...31, 0...31, ...] or :py:const:`crontab.ANY` for every Day of Month.
|
|
:type day_of_month: int, list, str
|
|
:param month: Month for execution. Either 0...12, [0...12, 0...12, ...] or :py:const:`crontab.ANY` for every Month.
|
|
:type month: int, list, str
|
|
:param day_of_week: Day of Week for execution. Either 0...6, [0...6, 0...6, ...] or :py:const:`crontab.ANY` for every Day of Week.
|
|
:type day_of_week: int, list, str
|
|
"""
|
|
if minute is not None:
|
|
self.minute = self.__conv_to_set__(minute)
|
|
if hour is not None:
|
|
self.hour = self.__conv_to_set__(hour)
|
|
if day_of_month is not None:
|
|
self.day_of_month = self.__conv_to_set__(day_of_month)
|
|
if month is not None:
|
|
self.month = self.__conv_to_set__(month)
|
|
if day_of_week is not None:
|
|
self.day_of_week = self.__conv_to_set__(day_of_week)
|
|
|
|
def __conv_to_set__(self, obj):
|
|
if obj is crontab.ANY:
|
|
return self.all_match()
|
|
elif isinstance(obj, (int, long) if sys.version_info < (3,0) else (int)):
|
|
return set([obj])
|
|
else:
|
|
return set(obj)
|
|
|
|
def __execution_needed_for__(self, minute, hour, day_of_month, month, day_of_week):
|
|
if self.__last_execution__ != [minute, hour, day_of_month, month, day_of_week]:
|
|
if minute in self.minute and hour in self.hour and day_of_month in self.day_of_month and month in self.month and day_of_week in self.day_of_week:
|
|
return True
|
|
return False
|
|
|
|
def __store_execution_reminder__(self, minute, hour, day_of_month, month, day_of_week):
|
|
self.__last_execution__ = [minute, hour, day_of_month, month, day_of_week]
|
|
|
|
def cron_execution(self, tm):
|
|
"""This Methods executes the Cron-Callback, if a execution is needed for the given time (depending on the parameters on initialisation)
|
|
|
|
:param tm: (Current) Time Value to be checked. The time needs to be given in seconds since 1970 (e.g. generated by int(time.time())).
|
|
:type tm: int
|
|
"""
|
|
if self.__last_cron_check_time__ is None:
|
|
self.__last_cron_check_time__ = tm - 1
|
|
#
|
|
for t in range(self.__last_cron_check_time__ + 1, tm + 1):
|
|
lt = time.localtime(t)
|
|
if self.__execution_needed_for__(lt[4], lt[3], lt[2], lt[1], lt[6]):
|
|
self.callback(self, *self.args, **self.kwargs)
|
|
self.__store_execution_reminder__(lt[4], lt[3], lt[2], lt[1], lt[6])
|
|
break
|
|
self.__last_cron_check_time__ = tm
|
|
|
|
def __init__(self, accuracy=30):
|
|
periodic.__init__(self, accuracy, self.__periodic__)
|
|
self.__crontab__ = []
|
|
|
|
def __periodic__(self, rt):
|
|
(rt)
|
|
tm = int(time.time())
|
|
for cronjob in self.__crontab__:
|
|
cronjob.cron_execution(tm)
|
|
|
|
def add_cronjob(self, minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs):
|
|
"""This Method adds a cronjob to be executed.
|
|
|
|
:param minute: Minute for execution. Either 0...59, [0...59, 0...59, ...] or :py:const:`crontab.ANY` for every Minute.
|
|
:type minute: int, list, str
|
|
:param hour: Hour for execution. Either 0...23, [0...23, 0...23, ...] or :py:const:`crontab.ANY` for every Hour.
|
|
:type hour: int, list, str
|
|
:param day_of_month: Day of Month for execution. Either 0...31, [0...31, 0...31, ...] or :py:const:`crontab.ANY` for every Day of Month.
|
|
:type day_of_month: int, list, str
|
|
:param month: Month for execution. Either 0...12, [0...12, 0...12, ...] or :py:const:`crontab.ANY` for every Month.
|
|
:type month: int, list, str
|
|
:param day_of_week: Day of Week for execution. Either 0...6, [0...6, 0...6, ...] or :py:const:`crontab.ANY` for every Day of Week.
|
|
:type day_of_week: int, list, str
|
|
:param callback: The callback to be executed. The instance of :py:class:`cronjob` will be given as the first, args and kwargs as the following parameters.
|
|
:type callback: func
|
|
|
|
.. note:: The ``callback`` will be executed with it's instance of :py:class:`cronjob` as the first parameter.
|
|
The given Arguments (:data:`args`) and keyword Arguments (:data:`kwargs`) will be stored in that object.
|
|
"""
|
|
self.__crontab__.append(self.cronjob(minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs))
|