Easily schedule multiple events and recurring tasks (simultaneously).
Project description
simple_scheduler
- Uses multiprocessing.
- The package uses a 24-hour clock, only.
- Simultaneously schedule any number of jobs.
- A job would be called any time within the 60-seconds[0-59] of a minute. This package does not begin execution of a job at the 0th-second of the minute.
Install
From PyPi :
pip install simple_scheduler
How to use?
Quick-start
Copy-paste the code at the bottom and run.
Long
There are two different schedulers:
>>> from simple_scheduler.event import event_scheduler
>>> from simple_scheduler.recurring import recurring_scheduler
Purpose of each scheduler:
>>> print(event_scheduler.__doc__)
Event occurs at an exact time.
e.g. scirpt_1 is called at 14:00 and 20:00
Each event is tried 3-times (but executed only once).
>>> print(recurring_scheduler.__doc__)
Recurring tasks are those that occur after every "x"-seconds.
(e.g. script_1 is called every 600 seconds)
Using only event_scheduler
>>> print(event_scheduler.add_job.__doc__)
Assigns an event to a process.
Parameters
----------
target : a callable function
tz : str
time zone (call the method .timezones() for more info)
when : list(str)
at what precise time(s) should the function be called
eg. ["12:34","23:45", ...] --> please "only" use 24-hour clock,
with ":" as separator
args : tuple(object,), optional
un-named argumets for the "target" callable
the default is ()
kwargs : dict{key:object}, optional
named argumets for the "target" callable
the default is {}
Raises
------
Exception
- If time (in "when"-list) is not a collection of "int:int"
eg. ["12:30am","2:30 pm", ...] --> please "only" use 24-hour clock,
with ":" as separator
Returns
-------
None.
correct argument precedence in a function
>>> def target(a, b=1, *args, **kwargs):
print(a, b, args, kwargs)
Add above target function twice. Each function would be called on timestamps (hour:minute) mentioned in list WHEN.
>>> WHEN = ["16:55", "16:56"]
>>> TZ = "Asia/Kolkata"
>>> event_scheduler.add_job(target= target,
args = (0,), # ... use "," for single arguments
kwargs = {"b":2},
when = WHEN,
tz = TZ)
>>> event_scheduler.add_job(target= target,
args = (0, 2, "arg1", "arg2"),
kwargs = {"key1":"value1",
"key2":"value2"},
when = WHEN,
tz = TZ)
>>> event_scheduler.run()
Wed Mar 17 16:55:32 2021 :: target [event @16:55 | Asia/Kolkata]
0 2 () {}
Wed Mar 17 16:55:32 2021 :: target [event @16:55 | Asia/Kolkata]
0 2 ('arg1', 'arg2') {'key1': 'value1', 'key2': 'value2'}
Wed Mar 17 16:56:27 2021 :: target [event @16:56 | Asia/Kolkata]
0 2 () {}
Wed Mar 17 16:56:27 2021 :: target [event @16:56 | Asia/Kolkata]
0 2 ('arg1', 'arg2') {'key1': 'value1', 'key2': 'value2'}
Using only recurrent_scheduler
>>> print(recurring_scheduler.add_job.__doc__)
Assigns an periodic task to a process.
Parameters
----------
target : a callable function
period_in_seconds : int
the time period in seconds to execute this function
args : tuple(object,), optional
un-named argumets for the "target" callable
the default is ()
kwargs : dict{key:object}, optional
named argumets for the "target" callable
the default is {}
Returns
-------
None.
>>> from time import sleep
>>> def wait(t):
sleep(t)
print(f"I waited {t} seconds")
>>> recurring_scheduler.add_job(target=wait,
kwargs={"t":10},
period_in_seconds=5)
>>> recurring_scheduler.run()
Wed Mar 17 17:14:30 2021 :: wait [recurring | 5-second(s)]
Wed Mar 17 17:14:35 2021 :: wait [recurring | 5-second(s)]
I waited 10 seconds
Wed Mar 17 17:14:40 2021 :: wait [recurring | 5-second(s)]
I waited 10 seconds
Wed Mar 17 17:14:45 2021 :: wait [recurring | 5-second(s)]
Note
The execution time of above "wait"-function is 10 seconds. But, is being called every 5-seconds.
Toggle verbose
>>> event_scheduler.verbose = False
>>> recurring_scheduler.verbose = True
Complete code:
from time import sleep
from simple_scheduler.event import event_scheduler
from simple_scheduler.recurring import recurring_scheduler
print(event_scheduler.__doc__)
print(event_scheduler.add_job.__doc__)
print()
print(recurring_scheduler.__doc__)
print(recurring_scheduler.add_job.__doc__)
print()
# Select your timezone
recurring_scheduler.timezones()
TZ = "Asia/Kolkata"
WHEN = ["17:26", "17:27"], # 1-min difference
# correct argument precedence in a function
def target(a, b=1, *args, **kwargs):
print(a, b, args, kwargs)
def wait(t):
sleep(t)
print(f"I waited {t} seconds")
event_scheduler.add_job(target= target,
args = (0,), # ... use "," for single arguments
kwargs = {"b":2},
when = WHEN,
tz = TZ)
event_scheduler.add_job(target= target,
args = (0, 2, "arg1", "arg2"),
kwargs = {"key1":"value1",
"key2":"value2"},
when = WHEN,
tz = TZ)
recurring_scheduler.add_job(target=wait,
kwargs={"t":10}, # The function-wait() executes in 10 seconds
period_in_seconds=5) # but is called every 5-seconds.
event_scheduler.run()
recurring_scheduler.run()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distributions
Close
Hashes for simple_scheduler-1.0.0b1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f5331be0a9a316635599fd9b1ce3b72ffbb9653c04a96cd8cffc07ad1fadd32d |
|
MD5 | 4157645da91859217653e66c86c05600 |
|
BLAKE2b-256 | 1a521792cd52750cff504fe0709d7c23fc1d79454acbe4c5ab8dbdf2f3188b02 |
Close
Hashes for simple_scheduler-1.0.0b1-py2.py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | cc6677202ca9e124868080a65b05947ef472121d429942b66c905f671a630f90 |
|
MD5 | 546355e65a71056a89fe374f7e92bbf3 |
|
BLAKE2b-256 | ead6202f67a771aa064f12006238c33f56d52228b168a09e3dbb7c86df9aa720 |