DigonIO/scheduler — 91 Stars

Topics

#datetime #library #pypi #python #python-library #python-scheduler #python3 #schedule #scheduler #timezone #timezones

Language breakdown

Python 99.8%

Shell 0.2%

▸ Files click to expand

README

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

If you find the scheduler library beneficial, please consider supporting the project by starring it on GitHub.

Features

Easy and user friendly in-process Job scheduling

(Quick Start)

Asyncio scheduler (Example)
Threading scheduler (Example)
Timezone compatibility (Example)
Passing of parameters

(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

Installation

pip

scheduler can be installed directly from the PyPI repositories with:

pip install scheduler

Alternatively install scheduler from the git repository with:

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

Example: How to schedule Jobs

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

[//]: # (This example is not directly included in the testing environment. Make sure to also update the corresponding test in tests/test_readme.py when updating the following example.)

import datetime as dt
from scheduler import Scheduler from scheduler.trigger import Monday, Tuesday
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(Monday(), foo) schedule.weekly(Monday(dt.time(hour=16, minute=30)), foo)
schedule.once(dt.timedelta(minutes=10), foo) schedule.once(Tuesday(), 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)

maxexec=inf, tzinfo=None, priorityfunction=linearpriorityfunction, #jobs=9
type     function / alias due at                 due in      attempts weight -------- ---------------- ------------------- --------- ------------- ------ MINUTELY foo()            2021-05-26 03:55:15   0:00:14         0/inf      1 CYCLIC   foo()            2021-05-26 04:05:00   0:09:59         0/inf      1 ONCE     foo()            2021-05-26 04:05:00   0:09:59           0/1      1 HOURLY   foo()            2021-05-26 04:30:15   0:35:14         0/inf      1 DAILY    foo()            2021-05-26 16:30:00  12:34:59         0/inf      1 WEEKLY   foo()            2021-05-31 00:00:00    4 days         0/inf      1 WEEKLY   foo()            2021-05-31 16:30:00    5 days         0/inf      1 ONCE     foo()            2021-06-01 00:00:00    5 days           0/1      1 ONCE     foo()            2022-02-15 00:45:00  264 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)

Resources

Sponsor

Digon.IO provides dev & data end-to-end consulting for SMEs and software companies. (Website) (Technical Blog)

The sponsor logo is the property of Digon.IO GmbH. Standard trademark and copyright restrictions apply to any use outside this repository.

License

Library source code: Licensed under LGPLv3.

🔗 More in this category