scheduler

A simple in-process python scheduler library, designed to be integrated seamlessly with the datetime standard library. Due to the support of datetime objects, scheduler is able to work with time zones. This implementation enables the planning of Job s depending on time cycles, fixed times, weekdays, dates, weights, offsets and execution counts.

Features

• Easy and user friendly in-process Job scheduling
  • Create recurring Jobs by given date, time, weekday, ...
  • Create recurring Jobs with a given timedelta
  • Oneshot Jobs
• Job prioritization with linear weighting
• datetime compatibility
• Timezone compatibility
• Lightweight
• Limit and track the Job execution count
• High test coverage
• Online documentation

Installation

scheduler can be installed using pip with the following command:

pip install git+https://gitlab.com/DigonIO/scheduler.git

Alternatively clone the repository and install with:

git clone REPLACE_ME
cd scheduler
pip install .

Example: How to schedule Jobs

Some basics are presented here. For advanced scheduling examples please visit the online documentation. The following example shows how the Scheduler is instantiated and how cyclic Jobs are created:

import time
import datetime as dt
from scheduler import Scheduler, Weekday

def foo(msg = "bar"):
    print(msg)

sch = Scheduler()

sch.schedule(foo, dt.timedelta(minutes=10))  # every 10 minutes
sch.schedule(foo, dt.time(hour=16, minute=45))  # every day at 16:45
sch.schedule(foo, Weekday.MONDAY)  # every monday at 00:00

# every monday at 16:45
sch.schedule(
    foo,
    (Weekday.MONDAY, dt.time(hour=16, minute=45)),
)

# every friday at 00:00, every 10 minutes and every monday at 16:45
sch.schedule(
    foo,
    [
        Weekday.FRIDAY,
        dt.timedelta(minutes=10),
        (Weekday.MONDAY, dt.time(hour=16, minute=45)),
    ],
)

Besides cyclic Jobs, oneshot Jobs can also be easily created:

sch.once(foo, dt.datetime(year=2021, month=2, day=11))  # at given datetime
sch.once(foo, dt.timedelta(minutes=10))  # in 10 minutes

Pass parameters to the function handle foo:

sch.once(foo, dt.timedelta(seconds=10000), params={"msg": "fizz"})
sch.schedule(foo, dt.timedelta(minutes=1), params={"msg": "buzz"})

Create a loop in the host program to execute pending Jobs:

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

Build the documentation

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

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

Testing

Testing is done using pytest. Using pytest-cov and coverage a report for the tests can be generated with:

pytest --cov=scheduler/ tests/
coverage html

To test the examples in the documentation run:

pytest --doctest-modules doc/examples.rst

TODO

• Features
  • Support of monthly recurring Jobs (e.g. every second Monday in June and October)
  • Add __repr__ methods to Job and Scheduler
  • Execute all scheduled Jobs
  • Delete all scheduled Jobs
  • Optional Job flag: Discard missed executions befor the last pending execution
  • Execute a Job until a certain datetime stamp
  • Thread safety and background tasks
• Documentation
  • Notes on performance
  • Comparison to APScheduler and schedule
  • where to get help
  • FAQ

License

This software is published under the GPLv3 license.