task (Task Module)¶

Author:

Dirk Alders <sudo-dirk@mount-mockery.de>

Description:

This Module supports helpfull classes for queues, tasks, …

Submodules:

task.crontab
task.delayed
task.periodic
task.queue
task.threaded_queue

Unittest:

See also the unittest documentation.

Module Documentation:

class task.crontab(accuracy=30)¶

Class to execute a callback at the specified time conditions. See also parent periodic.

Parameters:	accuracy (float) – Repeat time in seconds for background task checking event triggering. This time is the maximum delay between specified time condition and the execution.

Example:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import sys
sys.path.append('../..')
import task
import time


def print_localtime(cj):
    print(time.localtime())


ct = task.crontab(accuracy=7)
minute = int(time.strftime('%M'))
ct.add_cronjob([minute + 1, minute + 3], task.crontab.ANY, task.crontab.ANY, task.crontab.ANY, task.crontab.ANY, print_localtime)
print('Cronjob added for Minute: %02d, %02d\n--------------------------------\n' % (minute + 1, minute + 3))
ct.run()
try:
    time.sleep(195)
    ct.stop()
    ct.join()
finally:
    ct.stop()

Will result to the following output:

Cronjob added for Minute: 02, 04
--------------------------------

time.struct_time(tm_year=2023, tm_mon=10, tm_mday=26, tm_hour=0, tm_min=2, tm_sec=0, tm_wday=3, tm_yday=299, tm_isdst=1)
time.struct_time(tm_year=2023, tm_mon=10, tm_mday=26, tm_hour=0, tm_min=4, tm_sec=6, tm_wday=3, tm_yday=299, tm_isdst=1)

ANY = '*'¶: Constant for matching every condition.

add_cronjob(minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs)¶

This Method adds a cronjob to be executed.

Parameters:

minute (int, list, str) – Minute for execution. Either 0…59, [0…59, 0…59, …] or crontab.ANY for every Minute.
hour (int, list, str) – Hour for execution. Either 0…23, [0…23, 0…23, …] or crontab.ANY for every Hour.
day_of_month (int, list, str) – Day of Month for execution. Either 0…31, [0…31, 0…31, …] or crontab.ANY for every Day of Month.
month (int, list, str) – Month for execution. Either 0…12, [0…12, 0…12, …] or crontab.ANY for every Month.
day_of_week (int, list, str) – Day of Week for execution. Either 0…6, [0…6, 0…6, …] or crontab.ANY for every Day of Week.
callback (func) – The callback to be executed. The instance of cronjob will be given as the first, args and kwargs as the following parameters.

Note

The callback will be executed with it’s instance of cronjob as the first parameter. The given Arguments (args) and keyword Arguments (kwargs) will be stored in that object.

class cronjob(minute, hour, day_of_month, month, day_of_week, callback, *args, **kwargs)¶

Class to handle cronjob parameters and cronjob changes.

Parameters:

minute (int, list, str) – Minute for execution. Either 0…59, [0…59, 0…59, …] or crontab.ANY for every Minute.
hour (int, list, str) – Hour for execution. Either 0…23, [0…23, 0…23, …] or crontab.ANY for every Hour.
day_of_month (int, list, str) – Day of Month for execution. Either 0…31, [0…31, 0…31, …] or crontab.ANY for every Day of Month.
month (int, list, str) – Month for execution. Either 0…12, [0…12, 0…12, …] or crontab.ANY for every Month.
day_of_week (int, list, str) – Day of Week for execution. Either 0…6, [0…6, 0…6, …] or crontab.ANY for every Day of Week.
callback (func) – The callback to be executed. The instance of cronjob will be given as the first, args and kwargs as the following parameters.

Note

This class should not be used stand alone. An instance will be created by adding a cronjob by using crontab.add_cronjob().

class all_match¶: Universal set - match everything

cron_execution(tm)¶

This Methods executes the Cron-Callback, if a execution is needed for the given time (depending on the parameters on initialisation)

Parameters:	tm (int) – (Current) Time Value to be checked. The time needs to be given in seconds since 1970 (e.g. generated by int(time.time())).

set_trigger_conditions(minute=None, hour=None, day_of_month=None, month=None, day_of_week=None)¶

This Method changes the execution parameters.

Parameters:

minute (int, list, str) – Minute for execution. Either 0…59, [0…59, 0…59, …] or crontab.ANY for every Minute.
hour (int, list, str) – Hour for execution. Either 0…23, [0…23, 0…23, …] or crontab.ANY for every Hour.
day_of_month (int, list, str) – Day of Month for execution. Either 0…31, [0…31, 0…31, …] or crontab.ANY for every Day of Month.
month (int, list, str) – Month for execution. Either 0…12, [0…12, 0…12, …] or crontab.ANY for every Month.
day_of_week (int, list, str) – Day of Week for execution. Either 0…6, [0…6, 0…6, …] or crontab.ANY for every Day of Week.

class task.delayed(cycle_time, callback, *args, **kwargs)¶

Class to execute a callback a given time in the future. See also parent periodic.

Parameters:	time (float) – Delay time for execution of the given callback callback (callback) – Callback to be executed args (args) – Arguments to be given to callback kwargs (kwargs) – Keword Arguments to be given to callback

Example:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import sys
sys.path.append('../..')
import task
import time


def time_print(txt):
    sys.stdout.write(time.asctime() + ': ' + txt + '\n')


print("task.delayed example:\n---------------------")
dt = task.delayed(5, time_print, "A delayed hello!")
dt.run()
try:
    time_print("starting...")
    dt.join()
finally:
    dt.stop()

Will result to the following output:

task.delayed example:
---------------------
Thu Oct 26 00:04:39 2023: starting...
Thu Oct 26 00:04:44 2023: A delayed hello!

run()¶: This starts the timer for the delayed execution.

class task.periodic(cycle_time, callback, *args, **kwargs)¶

Class to execute a callback cyclicly.

Parameters:	cycle_time (float) – Cycle time in seconds – callback will be executed every cycle_time seconds callback (callback) – Callback to be executed args (args) – Arguments to be given to the callback kwargs (kwargs) – Keword Arguments to be given to callback

Note

The Callback will get this instance as first argument, followed by args und kwargs.

Example:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import sys
sys.path.append('../..')
import task
import time

task_num = 0


def time_print(txt):
    sys.stdout.write(time.asctime() + ': ' + txt + '\n')


def hello(rt, name):
    global task_num
    task_num += 1
    if task_num >= 5:
        rt.stop()
    tn = task_num
    time_print("(Task %d) Hello %s!" % (tn, name))
    time.sleep(3.8)
    time_print("(Task %d) Ende!" % (tn))


print("task.periodic example:\n----------------------")
pt = task.periodic(2, hello, "from periodic example")
pt.run()
try:
    time_print("starting...")
    pt.join()
finally:
    pt.stop()

Will result to the following output:

task.periodic example:
----------------------
Thu Oct 26 00:04:20 2023: starting...
Thu Oct 26 00:04:20 2023: (Task 1) Hello from periodic example!
Thu Oct 26 00:04:22 2023: (Task 2) Hello from periodic example!
Thu Oct 26 00:04:23 2023: (Task 1) Ende!
Thu Oct 26 00:04:24 2023: (Task 3) Hello from periodic example!
Thu Oct 26 00:04:25 2023: (Task 2) Ende!
Thu Oct 26 00:04:26 2023: (Task 4) Hello from periodic example!
Thu Oct 26 00:04:27 2023: (Task 3) Ende!
Thu Oct 26 00:04:28 2023: (Task 5) Hello from periodic example!

join()¶: This blocks till the cyclic task is terminated.

Note

Using join means that somewhere has to be a condition calling stop() to terminate. Otherwise task.join() will never return.

run()¶: This starts the cyclic execution of the given callback.

stop()¶: This stops the execution of any further task.

class task.queue(expire=True)¶

Class to execute queued callbacks.

Parameters:	expire (bool) – The default value for expire. See also `expire()`.

Example:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import sys
sys.path.append('../..')
import task
import time

task_num = 0


def time_print(txt):
    sys.stdout.write(time.asctime() + ': ' + txt + '\n')


def hello(rt, name):
    global task_num
    task_num += 1
    if task_num >= 5:
        rt.stop()
    tn = task_num
    time_print("(Task %d) Hello %s!" % (tn, name))
    time.sleep(3.8)
    time_print("(Task %d) Ende!" % (tn))


print("task.queue example:\n----------------------")
q = task.queue()
q.enqueue(5, hello, "from queue example (5)")
q.enqueue(6, hello, "from queue example (6)")
q.enqueue(4, hello, "from queue example (4)")
q.run()

Will result to the following output:

task.queue example:
----------------------
Thu Oct 26 00:04:44 2023: (Task 1) Hello from queue example (4)!
Thu Oct 26 00:04:48 2023: (Task 1) Ende!
Thu Oct 26 00:04:48 2023: (Task 2) Hello from queue example (5)!
Thu Oct 26 00:04:52 2023: (Task 2) Ende!
Thu Oct 26 00:04:52 2023: (Task 3) Hello from queue example (6)!
Thu Oct 26 00:04:56 2023: (Task 3) Ende!

clean_queue()¶: This Methods removes all jobs from the queue.

Note

Be aware that already running jobs will not be terminated.

enqueue(priority, callback, *args, **kwargs)¶

This enqueues a given callback.

Parameters:	priority (number) – The priority indication number of this task. The lowest value will be queued first. callback (callback) – Callback to be executed args (args) – Arguments to be given to callback kwargs (kwargs) – Keword Arguments to be given to callback

Note

Callback will get this instance as first argument, followed by args und kwargs.

expire()¶: This sets the expire flag. That means that the process will stop after queue gets empty.

run()¶: This starts the execution of the queued callbacks.

stop()¶: This sets the stop flag. That means that the process will stop after finishing the active task.

class task.threaded_queue(expire=False)¶

Class to execute queued callbacks in a background thread (See also parent queue).

Parameters:	expire (bool) – The default value for expire. See also `queue.expire()`.

Example:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import sys
sys.path.append('../..')
import task
import time

task_num = 0


def time_print(txt):
    sys.stdout.write(time.asctime() + ': ' + txt + '\n')


def hello(rt, name):
    global task_num
    task_num += 1
    if task_num >= 5:
        rt.stop()
    tn = task_num
    time_print("(Task %d) Hello %s!" % (tn, name))
    time.sleep(3.8)
    time_print("(Task %d) Ende!" % (tn))


print("task.threaded_queue example:\n-------------------------------")
tq = task.threaded_queue()
tq.enqueue(5, hello, "from queue example (5)")
tq.enqueue(6, hello, "from queue example (6)")
tq.enqueue(4, hello, "from queue example (4)")
tq.run()
try:
    time_print("starting...")
    tq.join()
finally:
    tq.stop()

Will result to the following output:

task.threaded_queue example:
-------------------------------
Thu Oct 26 00:04:28 2023: (Task 1) Hello from queue example (4)!
Thu Oct 26 00:04:28 2023: starting...
Thu Oct 26 00:04:31 2023: (Task 1) Ende!
Thu Oct 26 00:04:31 2023: (Task 2) Hello from queue example (5)!
Thu Oct 26 00:04:35 2023: (Task 2) Ende!
Thu Oct 26 00:04:35 2023: (Task 3) Hello from queue example (6)!
Thu Oct 26 00:04:39 2023: (Task 3) Ende!

join()¶: This blocks till the queue is empty.

Note

If the queue does not run dry, join will block till the end of the days.

run()¶: This starts the execution of the queued callbacks.

stop()¶: This sets the stop flag. That means that the process will stop after finishing the active task.