Simple Cron

Simplecron is a simple and lightweight Python library for scheduling tasks using cron-like syntax. It allows you to define jobs that run at specific intervals or times, making it easy to automate repetitive tasks in your applications.

This project was inspired by schedule. and improved to add:

• Asynchronous functionnalities
• Redis, SQLite, Postgres provider support

Creating a schedule

Default scheduler

Simplecron uses a default scheduler that is created when the library is imported. You can create jobs using the every function, which creates a new job instance and attaches it to the default scheduler.

from simplecron import base


def callback(job: base.Job, *args, **kwargs):
    print("Hello, World!", job)


base.every(1).second.do(callback)


while True:
    base.run_pending()
    time.sleep(1)

The same code can be achieved using the start_blocking function, which simplifies the blocking loop:

from simplecron import base

def callback(job: base.Job, *args, **kwargs):
    print("Hello, World!", job)

base.every(1).second.do(callback)
base.start_blocking()

By calling every, a new job is created. second attaches the unit of time to the job and finally do attaches the callback function.

Custom scheduler

You can also create your own scheduler by instantiating the BaseScheduler class. This can allow you to have multiple schedulers running concurrently, each with its own set of jobs.

from simplecron.base import BaseScheduler

s1 = BaseScheduler()
s2 = BaseScheduler()

# Adds a job to the first scheduler that runs every second
s1.every(1).second.do(lambda job: print("Scheduler 1:", job))

# Adds a job to the second scheduler that runs every 2 seconds
s2.every(2).seconds.do(lambda job: print("Scheduler 2:", job))

def main():
	while True:
		s1.run_pending_jobs()
		s2.run_pending_jobs()
		time.sleep(1)

if __name__ == "__main__":
	main()

Event Listeners

You can attach event listeners to a scheduler to listen for specific events. There are three main listeners:

• before - Triggered before a job is executed.
• after - Triggered after a job is executed.
• before_all - Triggered before all jobs are executed.

from simplecron.base import default_scheduler
from simplecron.utils import EventListenerEnum

default_scheduler.event_listener(EventListenerEnum.BEFORE_ALL, lambda scheduler: print("Before all jobs"))

The same can be achieved using the shortcut method before_all_events, after_events, and before_events:

from simplecron.base import default_scheduler

default_scheduler.before_all_events(lambda scheduler: print("Before all jobs"))
default_scheduler.before_events(lambda job: print("Before job:", job))
default_scheduler.after_events(lambda job: print("After job:", job))

Jobs

Cancelling

To cancel a job, it simply needs to return an instance of Cancel.

def callback(job: Job, *args, **kwargs):
	print("Hello, World!", job)
	return Cancel(job, reason="Some reason")  # This will cancel the job after it runs once

Types of jobs

Every second

default_scheduler.every(1).second.do(callback)

Every X second

default_scheduler.every(15).seconds.do(callback)

Every minute

default_scheduler.every(1).minute.do(callback)

Every X minutes

default_scheduler.every(15).minutes.do(callback)

Every hour

default_scheduler.every(1).hour.do(callback)

Every hours

default_scheduler.every(1).hours.do(callback)

Every day

When no specific time is provided, the job will run automatically at the start of the day (00:00). If you need to run the job at a specific time, you must use the at method to specify the time in 24-hour format (HH:MM).

default_scheduler.every(1).day.do(callback)

default_scheduler.every(1).day.at(datetime.time(12, 00)).do(callback)

Every days

default_scheduler.every(1).days.do(callback)

Every week

If no specific day and time is povided, the job will run automatically at the start of the week (Monday at 00:00). If you need to run the job at a specific day, you must use one of the properties monday, tuesday, wednesday, thursday, friday, saturday or sunday.

You can also use the at method to specify the time in 24-hour format (HH:MM).

default_scheduler.every(1).week.do(callback)

default_scheduler.every(1).week.at(datetime.time(12, 00)).do(callback)

Every X day

default_scheduler.every(1).monday.do(callback)
default_scheduler.every(1).tuesday.do(callback)
default_scheduler.every(1).wednesday.do(callback)
default_scheduler.every(1).thursday.do(callback)
default_scheduler.every(1).friday.do(callback)
default_scheduler.every(1).saturday.do(callback)
default_scheduler.every(1).sunday.do(callback)

Tags

Tags allow you to categorize and filter jobs based on specific labels. This can be useful for organizing jobs, applying actions to groups of jobs, or selectively running certain jobs based on their tags.

base.every(15, tag="my_tag").seconds.do(executor)

You can also attach event listeners to specific jobs matching a certain set of tags or criteria:

import time

from simplecron.base import Job, default_scheduler, logger
from simplecron.utils import EventListenerEnum


def executor(job: Job):
    logger.info("Executor called")


def event_before(job: Job):
    print("Before job:", job._tags)


default_scheduler.create_every(10, tag="my_tag").seconds.do(executor)

default_scheduler.with_event_listener(
    EventListenerEnum.BEFORE, 
	event_before, 
	for_tags=["my_tag"]
)

while True:
	default_scheduler.run_pending()
	time.sleep(1)

Providers

Providers are external services or modules that can be integrated with the scheduler to extend its functionality. They allow you to connect your scheduled jobs with various platforms, APIs, or other systems seamlessly.

Redis Database provider

The example below will save the details of the scheduler and the jobs that were runned in a Redis backend:

import time

from simplecron import base
from simplecron.base import Job, logger
from simplecron.providers import RedisDatabase


def executor(job: Job):
    logger.warning("Executor called")


base.default_scheduler.providers.attach(RedisDatabase())
base.every(15).seconds.do(executor)
base.every(30).seconds.do(executor)


while True:
    base.run_pending()
    time.sleep(1)