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
Job
s by given date, time, datetime, weekday, ... - Create recurring
Job
s with a given timedelta - Oneshot
Job
s - Passing of parameters to
Job
(Example)
- Create recurring
- Timezone compatibility (Example)
- Parallel
Job
execution withthreading
(Example) Job
prioritizationJob
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 Job
s 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 Job
s 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.