Advanced Usage

This part of documentation will introduce extra usage of Schedule Manager.

Task Information

We are able to get task information from Task object.

Check Activation

Task provides is_running property to check the activation of the task.

>>> from schedule_manager import Task
>>> task = Task(name="task", job=print, args=("Hello task",))
>>> task.is_running
False
>>> task.period_day_at("15:00:00").start()
>>> task.is_running
True

Job Next Run Time

Task provides next_run property to check when the job will be done next time .

>>> from schedule_manager import Task
>>> from datetime import datetime
>>> task = Task(name="task", job=print, args=("Hello task",))
>>> datetime.now()
datetime.datetime(2020, 8, 9, 12, 14, 10, 361827)
>>> task.period_day_at("15:00:00").start()
>>> task.next_run
datetime.datetime(2020, 8, 9, 15, 0, 0, 802553)

Task Behavior with ignore_skipped flag

The ignore_skipped flag is used to control the behavior if the job take more time than time interval configured to the task.

If ignore_skipped flag is set to True, Task will ignore the overdue work.

Let’s use examples for explanation.

Set ignore_skipped flag to True

We will use following python code for testing.

"""test_ignore_skipped_True.py"""

from schedule_manager import Task
import time

# A task will take 5 seconds to do the job
task = Task(job=time.sleep, args=(5,), ignore_skipped=True)

# Set time interval to 2 second
task.period(2)

task.start()

while True:
    # Show job next run time
    print(task.next_run.strftime("%H:%M:%S"))

    time.sleep(1)

We can see that the works at following time are skipped when ignore_skipped flag is set to True.

  • 19:34:00
  • 19:34:02
  • 19:34:06
  • 19:34:08
  • 19:34:12
  • 19:34:14

Because the task is busy at those time.

$ python test_ignore_skipped_True.py
19:33:58
19:33:58
19:33:58
19:33:58
19:33:58
19:33:58
19:34:04
19:34:04
19:34:04
19:34:04
19:34:04
19:34:04
19:34:10
19:34:10
19:34:10
19:34:10
19:34:10
19:34:10
19:34:16
19:34:16
19:34:16

Set ignore_skipped flag to False

Now we use following python code for another test.

"""test_ignore_skipped_False.py"""

from schedule_manager import Task
import time

# A task will take 5 seconds to do the job
task = Task(job=time.sleep, args=(5,), ignore_skipped=False)

# Set time interval to 2 second
task.period(2)

task.start()

while True:
    # Show job next run time
    print(task.next_run.strftime("%H:%M:%S"))

    time.sleep(1)

If ignore_skipped flag is set to False, We can see that all works are not skipped even those works are overdue works.

$ python test_ignore_skipped_False.py
19:37:39
19:37:39
19:37:39
19:37:39
19:37:39
19:37:39
19:37:41
19:37:41
19:37:41
19:37:41
19:37:41
19:37:41
19:37:43
19:37:43
19:37:43
19:37:43
19:37:43
19:37:43
19:37:45
19:37:45

Task Count

ScheduleManager provides count property to show how many tasks are registered in this manager.

>>> from schedule_manager import ScheduleManager
>>> manager = ScheduleManager()
>>> manager.register_task(job=print, args=("Task 1",))
Task<(Task-47844607c2ef4354903824cf1abc70be, initial daemon, None)>
>>> manager.register_task(job=print, args=("Task 2",))
Task<(Task-d512b84c33384a6b9bcb09c5f5f00207, initial daemon, None)>
>>> manager.register_task(job=print, args=("Task 3",))
Task<(Task-0cdeb2892d244e1fb489ab943a5a70af, initial daemon, None)>
>>> manager.count    # how many tasks are registered in this manager
3

TaskGroup also provides count property to show how many tasks we keep in the task group.

>>> from schedule_manager import ScheduleManager
>>> manager = ScheduleManager()
>>> task1 = manager.register_task(job=print, args=("Task 1",))
>>> task2 = manager.register_task(job=print, args=("Task 2",))
>>> task3 = manager.register_task(job=print, args=("Task 3",))
>>> task1.period_day_at("22:00:00").start()
>>> manager.pending_tasks.count
2