Croner is a lightweight, asynchronous scheduling service boilerplate built in Python. It leverages APScheduler for robust job scheduling and Loguru for elegant logging.
With its modular plugin architecture, adding new jobs is as easy as dropping a new file into the plugins/ directory! 🪄
- ⚡ Asynchronous Execution: Built on
asynciofor high performance and non-blocking job execution. - 🧩 Plugin Architecture: Isolated jobs. Drop a class in the
plugins/folder and it runs automatically. - ⏰ Flexible Triggers: Supports all APScheduler triggers:
CronTrigger,IntervalTrigger, etc. - 📝 Beautiful Logging: Colorized, structured logging right out of the box using Loguru.
- 🛡️ Fault Tolerant: One failing job won't crash your scheduler.
- 🐘 Database Ready: Built-in, high-performance PostgreSQL async connection pooling with
asyncpgand.envsupport. - 🧹 Linting & Pre-commit (
prek): Integrated with Ruff and Ty for automatic formatting, linting, and type-checking. - 🧪 Testing Ready: Automated testing on commit using
pytestandpytest-asyncioconfigured directly in the pre-commit hooks.
Croner/
├── clients/
│ └── pg_client.py # Async PostgreSQL client with pool management
├── config/
│ └── config.py # Configurations & python-decouple .env loader
├── core/
│ ├── base.py # Contains the CronJobBase class
│ └── loader.py # Auto-loads plugins from the plugins directory
├── plugins/ # 🌟 Drop your jobs here!
│ ├── cron_example_job.py
│ └── example_plugin.py
├── main.py # The engine that starts APScheduler & DB
├── pyproject.toml # Project configuration
└── README.md # You are here!
Creating a new job is simple! Just inherit from CronJobBase.
from core.base import CronJobBase
from loguru import logger
import asyncio
from apscheduler.triggers.interval import IntervalTrigger
from clients.pg_client import PGClient
from config.config import DBConfig
class MyAwesomeJob(CronJobBase):
def get_trigger(self) -> IntervalTrigger:
# Run every 5 seconds! ⏱️
return IntervalTrigger(seconds=5)
def is_disabled(self) -> bool:
# Set to True to pause this job 🛑
return False
async def execute(self):
# Your awesome async code goes here! 🚀
logger.info("Look at me, I'm executing asynchronously!")
# Access the database easily!
client = PGClient.connect(DBConfig.MAIN_DB)
result = await client.fetchval("SELECT 1 AS status")
logger.info(f"Database says: {result}")-
Configuration (
.env) Create a.envfile in the root directory and set yourDB_URL:DB_URL=postgresql://user:pass@localhost:5432/main
-
Install dependencies Sync your environment using
uvand install the git hooks. (Requires Python 3.14+)uv sync uv run prek install
-
Start the service locally
uv run main.py
You can easily containerize the service using the provided Dockerfile.
# 1. Build the image
docker build -t croner .
# 2. Run the container passing your .env file
docker run --env-file .env cronerWatch as your plugins are automatically discovered and scheduled! 🎉