task package¶

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)¶

Bases: periodic

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=2025, tm_mon=11, tm_mday=27, tm_hour=0, tm_min=2, tm_sec=3, tm_wday=3, tm_yday=331, tm_isdst=0)
time.struct_time(tm_year=2025, tm_mon=11, tm_mday=27, tm_hour=0, tm_min=4, tm_sec=2, tm_wday=3, tm_yday=331, tm_isdst=0)

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)¶

Bases: object

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¶

Bases: set

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)¶

Bases: periodic

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 Nov 27 00:04:15 2025: starting...
Thu Nov 27 00:04:20 2025: A delayed hello!

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

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

Bases: object

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 Nov 27 00:04:20 2025: starting...
Thu Nov 27 00:04:20 2025: (Task 1) Hello from periodic example!
Thu Nov 27 00:04:22 2025: (Task 2) Hello from periodic example!
Thu Nov 27 00:04:24 2025: (Task 1) Ende!
Thu Nov 27 00:04:24 2025: (Task 3) Hello from periodic example!
Thu Nov 27 00:04:26 2025: (Task 2) Ende!
Thu Nov 27 00:04:26 2025: (Task 4) Hello from periodic example!
Thu Nov 27 00:04:28 2025: (Task 3) Ende!
Thu Nov 27 00:04:28 2025: (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)¶

Bases: object

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 Nov 27 00:04:40 2025: (Task 1) Hello from queue example (4)!
Thu Nov 27 00:04:44 2025: (Task 1) Ende!
Thu Nov 27 00:04:44 2025: (Task 2) Hello from queue example (5)!
Thu Nov 27 00:04:47 2025: (Task 2) Ende!
Thu Nov 27 00:04:47 2025: (Task 3) Hello from queue example (6)!
Thu Nov 27 00:04:51 2025: (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.

class job(priority, callback, *args, **kwargs)¶

Bases: object

run(queue)¶

qsize()¶

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)¶

Bases: queue

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 Nov 27 00:04:28 2025: (Task 1) Hello from queue example (4)!
Thu Nov 27 00:04:28 2025: starting...
Thu Nov 27 00:04:32 2025: (Task 1) Ende!
Thu Nov 27 00:04:32 2025: (Task 2) Hello from queue example (5)!
Thu Nov 27 00:04:36 2025: (Task 2) Ende!
Thu Nov 27 00:04:36 2025: (Task 3) Hello from queue example (6)!
Thu Nov 27 00:04:40 2025: (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.

task package¶

task (Task Module)¶

Table of Contents

Previous topic

Next topic

This Page