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 in order to attach multiple event listeners to the scheduler:

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

Async Jobs

Simplecron supports asynchronous job execution, allowing you to run jobs concurrently without blocking the main thread.

You just need to define your job functions as asynchronous functions using the async def syntax.

import asyncio

from simplecron import base


async def executor(*args, **kwargs):
    print("Executed!")


async def main():
    base.every(5).seconds.do(executor)
    while True:
        base.run_pending()
        await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())

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

Limiting runs

You can limit the number of times a job runs by using the with_limited_runs method. This is useful when you want a job to execute only a specific number of times before being automatically cancelled.

job = base.every(10).seconds.do(executor)
job.with_limited_runs(5)

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)

Example Usage

Monitoring a page every 5 minutes

The following example demonstrates how to monitor a web page every 5 minutes using Simplecron and Playwright:

from playwright.sync_api import Page, sync_playwright

from simplecron import base
from simplecron.base import Job, logger
from simplecron.context import Context


def monitor_page(job: Job, context: Context | None = None, **kwargs):
    page: Page = context.json_data.get("page")
    if page is not None:
        page.reload()
    logger.info("Page monitored...")


base.every(30).seconds.do(monitor_page)


def main():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()

        page.goto("https://example.com")
        base.start_blocking(context={"page": page})

        browser.close()


if __name__ == "__main__":
    main()

Sending batch emails

import time
from simplecron import base
from simplecron.base import Job, logger
from simplecron.providers import EmailProvider

def send_batch_emails(job: Job):
    logger.info("Sending batch emails...")

base.default_scheduler.providers.attach(EmailProvider())
base.every(10).minutes.do(send_batch_emails)

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

Dockerizing the scheduler

To dockerize the scheduler, you need to create a Dockerfile that sets up the Python environment and runs your scheduler script. The example below demonstrates how to do this using uv:

FROM python:3.14-slim

# Copy the project into the image
COPY . /app

# Disable development dependencies
ENV UV_NO_DEV=1

# Sync the project into a new environment, asserting the lockfile is up to date
WORKDIR /app
RUN uv sync --locked

# Presuming there is a `scheduler` command provided by the project
CMD ["uv", "run", "scheduler"]

import time
from simplecron import base
from simplecron.base import Job, logger

def simple_function(job: Job):
    logger.info("Executing simple function...")

base.every(10).minutes.do(simple_function)

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