hjortlund / scheduler Goto Github PK

A simple in-process python scheduler library with timezone and threading support. Schedule tasks by their time cycles, fixed times, weekdays, dates, weights, offsets and execution counts and automate Jobs.

• Easy and user friendly in-process Job scheduling (Quick Start)
  • Create recurring Jobs by given date, time, datetime, weekday, ...
  • Create recurring Jobs with a given timedelta
  • Oneshot Jobs
  • Passing of parameters to Job (Example)
• Timezone compatibility (Example)
• Parallel Job execution with threading (Example)
• Job prioritization
  • Default linear prioritization (Example)
  • User definable prioritization functions (Guide)
• Job tagging (Example)
• Job batching (Example)
• Job metadata (Example)
• Lightweight
• High test coverage
• Online documentation

scheduler can be installed using pip with the following command:

pip install scheduler

Alternatively clone the repository and install with:

git clone https://gitlab.com/DigonIO/scheduler.git
cd scheduler
pip install .

The following example shows how the Scheduler is instantiated and how basic Jobs are created. For advanced scheduling examples please visit the online documentation.

import datetime as dt
from scheduler import Scheduler
import scheduler.trigger as trigger

def foo():
    print("foo")

schedule = Scheduler()

schedule.cyclic(dt.timedelta(minutes=10), foo)

schedule.minutely(dt.time(second=15), foo)
schedule.hourly(dt.time(minute=30, second=15), foo)
schedule.daily(dt.time(hour=16, minute=30), foo)
schedule.weekly(trigger.Monday(), foo)
schedule.weekly(trigger.Monday(dt.time(hour=16, minute=30)), foo)

schedule.once(dt.timedelta(minutes=10), foo)
schedule.once(trigger.Monday(), foo)
schedule.once(dt.datetime(year=2022, month=2, day=15, minute=45), foo)

A human readable overview of the scheduled jobs can be created with a simple print statement:

print(schedule)

max_exec=inf, tzinfo=None, priority_function=linear_priority_function, #jobs=9

type     function         due at                 due in      attempts weight
-------- ---------------- ------------------- --------- ------------- ------
MINUTELY bar(..)          2021-06-18 00:37:15   0:00:14         0/inf      1
CYCLIC   foo()            2021-06-18 00:46:58   0:09:58         0/inf      1
ONCE     foo()            2021-06-18 00:46:59   0:09:58           0/1      1
HOURLY   foo()            2021-06-18 01:30:15   0:53:14         0/inf      1
DAILY    bar(..)          2021-06-18 16:30:00  15:52:59         0/inf      1
WEEKLY   foo()            2021-06-21 00:00:00    2 days         0/inf      1
ONCE     bar(..)          2021-06-21 00:00:00    2 days           0/1      1
WEEKLY   bar(..)          2021-06-21 16:30:00    3 days         0/inf      1
ONCE     foo()            2022-02-15 00:45:00  242 days           0/1      1

Executing pending Jobs periodically can be achieved with a simple loop:

import time

while True:
    schedule.exec_jobs()
    time.sleep(1)

The API documentation can either be viewed online or generated using Sphinx with numpydoc formatting. To build, run:

sphinx-build -b html doc/ doc/_build/html

Testing is done using pytest. With pytest-cov and coverage a report for the test coverage can be generated:

pytest --cov=scheduler/ tests/
coverage html

To test the examples in the documentation run:

pytest --doctest-modules doc/pages/*/*

This software is published under the LGPLv3 license.