Easily schedule single or recurring sync/async tasks
Project description
EasySchedule
Get Started
pip install easyschedule
import asyncio
from easyschedule.scheduler import EasyScheduler
scheduler = EasyScheduler()
default_args = {'args': [1, 2, 3]}
every_minute = '* * * * *'
every_5_minutes_wk_end = '*/5, * * * FRI,SAT'
@scheduler(schedule=every_minute, default_args=default_args)
def print_stuff(a, b, c):
print(f"a {a} b: {b} c {c}")
def print_more_stuff():
print_stuff(3,4,5)
async def main():
# start scheduler in background
sched = asyncio.create_task(scheduler.start())
await asyncio.sleep(10)
# add task after scheduler has already started
scheduler.schedule(print_more_stuff, schedule=cron2)
await sched
asyncio.run(main())
03-09 10:35:32 EasyScheduler WARNING weekday_stuff next_run_time: 2021-03-09 10:36:00.843493
03-09 10:35:42 EasyScheduler WARNING weekend_stuff next_run_time: 2021-03-13 00:31:00.853770
Cron syntax Compatability
EasySchedule is capable of parsing most cron schedule syntax
Monthly
First of month at 11:00 PM
0 23 1 * *
Daily
Every 2 Hours
0 */2 * *
Weekends Only
Every Hour Between 5:30 PM - 5:30 AM ##
30 17-23,0-5 * * SAT,SUN
Cron Generator
An easy & interactive way to build a cron schedule is available via crontab.guru
Note: unsupported syntax (currently)
@(non-standard)
@hourly
@daily
@anually
Scheduluing Single Tasks
EasySchedule is complete with single task scheduling
Usage with 'once' decorator
from datetime import datetime, timedelta
next_year = datetime.now() + timedelta(days=365)
@scheduler.once(date=next_year)
async def future_task():
## future work
pass
# current month: 2021-03-13 00:00:00
@scheduler.once(date_string='2021-04-13 00:00:00')
async def run_at_date():
## future work
pass
# current month: 2021-03-13 00:00:00
@scheduler.once(delta=timedelta(days=3))
async def run_after_delta():
## future work
pass
now_args={'kwargs': {'work': "Lots of work"}}
@scheduler.once(now=True, default_args=now_args)
async def run_now(work):
## future work
print(f"starting {work}")
pass
Schedule a task at or near application startup
notify = {
'kwargs': { 'emails': ['admin@company.org'] }
}
@scheduler.delayed_start(delay_in_seconds=30, default_args=notify)
async def notify_online(emails: str):
message = f"server is operational"
await send_emails(message, emails)
#something else
async def get_data():
return await requests.get('http://data-source')
@scheduler.startup()
async def update_database():
data = await get_data()
await db.update(data)
#something else
Schedule a task to run at application shutdown
notify = {
'kwargs': { 'emails': ['admin@company.org'] }
}
@scheduler.shutdown(default_args=notify)
async def notify_shutdown(emails: str):
message = f"server is shutting down"
await send_emails(message, emails)
#something else?
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
No source distribution files available for this release.See tutorial on generating distribution archives.
Built Distribution
Close
Hashes for easyschedule-0.100-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4ff5e15f0481343d560744750e5c8e8c0ce4ad57736408f675a2a75a56f02c45 |
|
MD5 | 392ad8245527ec4dfe36aaa5244b3cfa |
|
BLAKE2b-256 | b92816a1b91744c9cf9ac6595569e962c19e5b1d96376b4bcd88fa3e39098d99 |