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
Job
s by given date, time, weekday, ... - Create recurring
Job
s with a given timedelta - Oneshot
Job
s
- Create recurring
Job
prioritization with linear weightingdatetime
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 Job
s 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 Job
s, oneshot Job
s 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 Job
s:
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
Job
s (e.g. every second Monday in June and October) - Add
__repr__
methods toJob
andScheduler
- Execute all scheduled
Job
s - Delete all scheduled
Job
s - Optional
Job
flag: Discard missed executions befor the last pending execution - Execute a
Job
until a certain datetime stamp - Thread safety and background tasks
- Support of monthly recurring
- Documentation
- Notes on performance
- Comparison to APScheduler and schedule
- where to get help
- FAQ
License
This software is published under the GPLv3 license.