A Python background task scheduler with cron-like clock-boundary execution, per-task dependency isolation and SQLite-backed run tracking. Tasks are plain *.py files with YAML frontmatter and a start() entry point — discovered and hot-reloaded without restarts.
Super simple example task:
"""
---
title: "Hello World Task"
description: "A simple example task"
dependencies: []
enabled: true
timeout: 60
---
"""
from task_scheduler.decorators import repeat, every
from helpers import log_task_start, log_task_complete, log_info
@repeat(every(10).minutes)
def start():
"""This function will be called every 10 minutes at clock boundaries"""
log_info("Hello from scheduled task!") # Logs to logs/tasks.log with [Hello World Task] prefixComes with more built in example tasks to get you started.
- Cron-like Scheduling: Tasks run at clock boundaries (e.g., every 5 minutes at :00, :05, :10) instead of relative intervals
- Automatic Dependency Management: Tasks specify their own Python package requirements in frontmatter
- Dynamic Task Loading: Automatically discovers and loads new task files without restart
- Flexible Configuration Management: Centralized config system with task-specific configs and secrets management
- Memory Management: Built-in memory cleanup and monitoring to prevent memory leaks
- Database Tracking: SQLite database tracks execution history and handles missed schedules
- Comprehensive Logging: Structured logging with rotation and task-specific logs
- Hybrid Scheduling System: Combines cron-like precision with overdue task detection
- Frontmatter Task Configuration: YAML frontmatter in docstrings for IDE compatibility
- Hot Configuration Reloading: Config changes applied automatically without restart
High level overview of project structure:
task-scheduler/
├── task_scheduler/ # Core scheduler package
├── tasks/ # Directory for task files
├── helpers/ # Reusable helper modules with dependencies
│ ├── __init__.py # Helper module imports
│ ├── config_loader.py # Configuration management
│ ├── system_notifications.py # Cross-platform notifications
│ ├── external_execution.py # External script/command execution
│ ├── task_logging.py # Task logging utilities
│ └── ... # More helper modules
├── config/ # Configuration files
│ └── config.yaml
├── logs/ # Log files
├── data/ # Database and data files
├── scripts/ # Setup and utility scripts
│ ├── install.sh # Installation script (Linux only)
│ ├── setup_cron.sh # Cron setup script (Linux/macOS)
│ ├── run_scheduler.sh # Wrapper script for cron (auto-generated from setup_cron.sh)
│ ├── reset_schedules.py # Database schedule reset utility
│ ├── restart_scheduler.py # Advanced process management script
│ ├── rotate_logs.sh # Log rotation script (auto-generated from setup_cron.sh)
│ ├── monitor_scheduler.sh # Automatic scheduler monitoring and restart (auto-generated from setup_cron.sh)
│ └── run_tests.py # Test runner script
├── tests/ # Pytest test suite
│ ├── ... # Test files
│ └── README.md # Testing documentation
├── main.py # Main entry point
├── requirements.txt # Python dependencies
├── pytest.ini # Pytest configuration
└── README.md
-
Clone or download the project:
git clone https://github.com/kinged007/py-task-scheduler cd py-task-scheduler -
Run the installation script (requires root):
sudo ./scripts/install.sh
-
Start the service:
sudo systemctl start task-scheduler sudo systemctl status task-scheduler
-
Setup the project:
git clone https://github.com/kinged007/py-task-scheduler cd py-task-scheduler python3 -m venv venv source venv/bin/activate pip install -r requirements.txt
-
Run the cron setup script:
./scripts/setup_cron.sh
This creates:
monitor_scheduler.sh- Automatic scheduler monitoring and restartrestart_scheduler.py- Advanced process management scriptrotate_logs.sh- Log rotation script
-
Follow the instructions to add cron entries for automatic monitoring.
-
Install dependencies:
python3 -m venv venv source venv/bin/activate pip install -r requirements.txt -
Run the scheduler:
python main.py --config config/config.yaml
Tasks are Python files placed in the tasks/ directory. Each task file includes YAML frontmatter to specify dependencies and metadata, and a start() function to define the task logic.
The task system supports YAML frontmatter at the top of the task file, either wrapped in a docstring or as a standalone block:
"""
---
title: "Hello World Task"
description: "A simple example task"
dependencies:
- "requests>=2.25.0"
- "beautifulsoup4==4.9.3"
enabled: true
timeout: 300
---
"""
from task_scheduler.decorators import repeat, every
import requests
@repeat(every(10).minutes)
def start():
"""This function will be called every 10 minutes at clock boundaries"""
print("Hello from scheduled task!")
# Your task logic here
response = requests.get("https://httpbin.org/json")
print(f"Response status: {response.status_code}")---
title: "Hello World Task"
description: "A simple example task"
dependencies:
- "requests>=2.25.0"
- "beautifulsoup4==4.9.3"
enabled: true
timeout: 300
---
from task_scheduler.decorators import repeat, every
import requests
@repeat(every(10).minutes)
def start():
"""This function will be called every 10 minutes at clock boundaries"""
print("Hello from scheduled task!")
# Your task logic here
response = requests.get("https://httpbin.org/json")
print(f"Response status: {response.status_code}")Note: The Python-syntax-correct format (Option 1) is recommended as it allows the task files to be valid Python modules that can be imported and tested independently, as well as pass python linters in IDEs.
The frontmatter uses YAML format and supports the following fields:
| Field | Type | Default | Description |
|---|---|---|---|
title |
string | filename | Task display name |
description |
string | "" | Task description |
dependencies |
list | [] | Python packages to install (pip format) |
python_version |
string | "3.8" | Minimum Python version |
enabled |
boolean | true | Whether task is active |
timeout |
integer | 300 | Task timeout in seconds |
retry_count |
integer | 0 | Number of retries on failure |
retry_delay |
integer | 60 | Delay between retries in seconds |
Dependencies follow standard pip requirement format:
dependencies:
- "requests>=2.25.0" # Minimum version
- "beautifulsoup4==4.9.3" # Exact version
- "pandas>=1.3.0,<2.0.0" # Version range
- "python-dateutil>=2.8.0" # Package with hyphens
- "plyer>=2.1.0" # Cross-platform notificationsThe task parser validates:
- ✅ YAML syntax: Must be valid YAML format
- ✅ Required fields:
titleanddescriptionare recommended - ✅ Data types: Fields must match expected types
- ✅ Python syntax: When using docstring format, file must be valid Python
Edit config/config.yaml to customize the scheduler behavior:
scheduler:
loop_interval: 10 # Main loop interval (seconds)
task_check_interval: 5 # Task file scan interval (seconds)
memory_cleanup_interval: 300 # Memory cleanup interval (seconds)
max_memory_usage: 500 # Max memory usage in MB before restart
database:
path: "data/scheduler.db"
backup_interval: 3600 # Database backup interval (seconds)
logging:
level: "INFO" # Log level (DEBUG, INFO, WARNING, ERROR)
file: "scheduler.log"
max_size: "10 MB"
retention: "7 days"
rotation: "daily"
virtual_env:
path: "venv"
python_executable: "python3"
tasks:
directory: "tasks"
file_pattern: "*.py"
reload_on_change: true
include_example_tasks: false # Include tasks prefixed with 'example_'The scheduler includes several example tasks (prefixed with example_) that demonstrate various features:
example_hello_world.py- Basic task with simple loggingexample_web_scraper.py- Web scraping with requestsexample_daily_cleanup.py- Daily maintenance operationsexample_system_monitor.py- System monitoring and alertsexample_external_execution.py- External script executionexample_system_notification.py- Cross-platform notificationsexample_config_usage.py- Configuration management and task-specific configsexample_helper_showcase.py- Comprehensive helper system demonstration
Configuration: Set include_example_tasks: true in config/config.yaml to enable example tasks, or false to ignore them. This is useful for:
- Development: Enable examples to test scheduler functionality
- Production: Disable examples to run only your custom tasks
- Learning: Study example implementations for best practices
The scheduler includes a centralized configuration system that allows tasks to load their own configuration files for secrets and task-specific settings.
Use the config helper to manage configurations in your tasks with simple imports:
# Simple import from helpers package
from helpers import load_config, get_config_value, save_config, log
# Load main configuration
config = load_config() # Loads config/config.yaml
# Load task-specific configuration
task_config = load_config("my_task.yaml") # Loads config/my_task.yaml
# Get specific values with dot notation and defaults
api_key = get_config_value("api.key", "secrets.yaml", default="not-set")
timeout = get_config_value("timeout", "my_task.yaml", default=30)
# Save configuration
new_config = {"setting": "value"}
save_config(new_config, "my_task.yaml")Available Functions:
load_config(config_name)- Load any YAML config from config/ directoryget_config_value(key_path, config_name, default)- Get values using dot notationsave_config(config_data, config_name)- Save configuration datalist_config_files()- List all available config filesget_project_root()- Get the project root directory
- Copy the example:
cp config/secrets.yaml.example config/secrets.yaml - Add your secrets to
config/secrets.yaml - Load in tasks:
secrets = load_config("secrets.yaml") api_key = secrets.get("api_keys", {}).get("openai")
The secrets.yaml file is automatically excluded from git commits.
The scheduler includes a comprehensive helper system that provides reusable functions for common task operations. All helpers are available through simple imports:
# Import everything you need from the helpers package
from helpers import (
# Configuration management
load_config,
get_config_value,
save_config,
# Logging
log,
log_info,
log_error,
log_warning,
# System notifications
send_notification,
send_and_log_notification,
# External execution
execute_python_script,
execute_cli_command,
)-
Configuration Management (
config_loader.py):- Centralized config loading with task-specific configs
- Secrets management and dot notation access
- Automatic fallbacks and error handling
-
Task Logging (
task_logging.py):- Structured logging with automatic task name detection
- Multiple log levels and task-specific log files
- Execution result logging and error handling
-
System Notifications (
system_notifications.py):- Cross-platform desktop notifications
- Integration with logging system
- Configurable notification settings
-
External Execution (
external_execution.py):- Execute Python scripts and CLI commands
- Retry mechanisms and timeout handling
- Result processing and error capture
Each helper manages its own dependencies automatically, so you only install what you use.
# Run scheduler
python main.py
# Run with custom config
python main.py --config /path/to/config.yaml
# Run as daemon (Linux only)
python main.py --daemon
# Show status
python main.py --statusThe scheduler includes advanced process management features for production use:
# Restart the scheduler (graceful shutdown + start)
python scripts/restart_scheduler.py
# Restart with custom timeout
python scripts/restart_scheduler.py --timeout 15
# Force restart if graceful shutdown fails
python scripts/restart_scheduler.py --force
# Check what would happen without making changes
python scripts/restart_scheduler.py --dry-run
# Start as daemon process
python scripts/restart_scheduler.py --daemonThe scheduler automatically detects and applies configuration changes without restart:
- Config file monitoring: Checks
config/config.yamlevery 5 seconds - Memory limit updates: Changes to
max_memory_usageare applied immediately - Directory changes: Task directory changes are detected and applied
The scheduler runs with a recognizable process name for easy management:
# Check if scheduler is running
ps aux | grep myautohub-scheduler
# Find scheduler process ID
pgrep -f myautohub-scheduler
# Using built-in script
python scripts/restart_scheduler.py --dry-run# Start service
sudo systemctl start task-scheduler
# Stop service
sudo systemctl stop task-scheduler
# Restart service
sudo systemctl restart task-scheduler
# Check status
sudo systemctl status task-scheduler
# View logs
sudo journalctl -u task-scheduler -f
# Enable auto-start on boot
sudo systemctl enable task-schedulerUse the monitoring script for automatic scheduler management:
# Check and start scheduler if not running
bash scripts/monitor_scheduler.sh
# Set up cron job for automatic monitoring (every 5 minutes)
*/5 * * * * /path/to/scripts/monitor_scheduler.shThe monitoring script:
- Detects if scheduler is running using process name
- Automatically starts scheduler if stopped
- Uses the restart script for reliable startup
- Logs all activities to
logs/monitor.log
logs/scheduler.log- Main scheduler logslogs/tasks.log- Task execution logslogs/errors.log- Error logs onlylogs/monitor.log- Monitoring script logs
The SQLite database (data/scheduler.db) contains:
- Task execution history
- Next run times
- Scheduler state
The scheduler automatically monitors memory usage and will restart if it exceeds the configured limit.
The scheduler uses a hybrid cron-like scheduling system that combines precision timing with overdue task detection.
Tasks run at clock boundaries instead of relative intervals:
from task_scheduler.decorators import repeat, every
# Every N minutes - runs at clock boundaries
@repeat(every(5).minutes) # Runs at :00, :05, :10, :15, :20, etc.
@repeat(every(10).minutes) # Runs at :00, :10, :20, :30, :40, :50
@repeat(every(30).minutes) # Runs at :00, :30
# Every N hours - runs at hour boundaries
@repeat(every(2).hours) # Runs at 00:00, 02:00, 04:00, etc.
@repeat(every(6).hours) # Runs at 00:00, 06:00, 12:00, 18:00
# Daily tasks - runs at specific times
@repeat(every(1).day) # Runs at 00:00 dailyThe scheduler uses the schedule library for traditional scheduling patterns.
For more complex scheduling patterns:
# At specific times (uses traditional schedule library)
@repeat(every().day.at("10:30")) # Daily at 10:30 AM
@repeat(every().monday.at("09:00")) # Every Monday at 9:00 AM
@repeat(every().hour.at(":15")) # 15 minutes past every hour- ✅ Predictable timing: Tasks run at exact clock boundaries
- ✅ System alignment: Multiple instances stay synchronized
- ✅ Resource efficiency: Avoids drift and timing conflicts
- ✅ Monitoring friendly: Easy to predict when tasks will run
The project includes a comprehensive pytest test suite to ensure reliability and functionality.
# Run all tests
python scripts/run_tests.py
# Run specific test file
python scripts/run_tests.py tests/test_installation.py
# Run with coverage reporting
python scripts/run_tests.py --coverage
# Skip slow tests (for quick validation)
python scripts/run_tests.py --fast
# Direct pytest usage
pytest -v
pytest -m "not slow"For detailed testing documentation, see tests/README.md.
The reset_schedules.py utility allows you to reset task schedules to clear overdue backlogs or prepare for testing:
# Preview what would be reset (dry run)
python scripts/reset_schedules.py --dry-run
# Reset only overdue tasks to future times
python scripts/reset_schedules.py --future-only
# Reset specific task
python scripts/reset_schedules.py --task example_hello_world
# Reset multiple specific tasks
python scripts/reset_schedules.py --task task1 --task task2
# Reset all tasks
python scripts/reset_schedules.py --allUse Cases:
- Clear overdue task backlog after system downtime
- Reset schedules for development/testing
- Reschedule tasks after maintenance windows
- Fix problematic task schedules
The script intelligently calculates next run times based on each task's schedule configuration and maintains proper clock boundary alignment.
- Overdue tasks now properly execute when detected
- Fixed job matching logic for CronLikeJob objects
- Comprehensive testing infrastructure
- Smart schedule reset tool with dry-run support
- Moved tests to dedicated directory with proper organization
- Updated README with testing and database management sections
- Initial release with core functionality
- Cron-like scheduling system
- YAML frontmatter for task metadata
- Automatic dependency management
- Dynamic task loading and unloading
- Memory management and cleanup
- SQLite database for execution tracking
- Virtual environment isolation
- Comprehensive logging system
- CLI and systemd service support
- Task not running: Check logs for errors, verify task file syntax
- Dependencies not installing: Check virtual environment permissions
- High memory usage: Reduce
memory_cleanup_intervalormax_memory_usage - Tasks not detected: Verify file is in
tasks/directory with.pyextension and YAML frontmatter - Overdue tasks not executing: Use
python scripts/reset_schedules.py --dry-runto check schedules - Test failures: Run
python scripts/run_tests.py --verbosefor detailed test output
Enable debug logging in config/config.yaml:
logging:
level: "DEBUG"Test a task manually:
cd /path/to/task-scheduler
source venv/bin/activate
# Method 1: Direct import (works with Python-syntax frontmatter)
python -c "
import sys
sys.path.append('.')
import tasks.your_task_file as task
task.start()
"
# Method 2: Test individual functions
python -c "
import sys
sys.path.append('.')
from tasks.example_system_notification import send_notification
result = send_notification('Test', 'Hello from manual test!')
print(f'Notification result: {result}')
"
# Method 3: Syntax validation
python -m py_compile tasks/your_task_file.py
# Method 4: Run automated tests
python scripts/run_tests.py tests/test_installation.py- Run as dedicated user (not root)
- Restrict file permissions on task files
- Validate task file contents before deployment
- Monitor log files for suspicious activity
- Use virtual environment isolation
The task scheduler includes a modular helper system for common operations. Helper modules are located in the helpers/ directory and have their own dependencies specified in YAML frontmatter.
Cross-platform system notification utilities:
from helpers import send_notification, send_and_log_notification
# Send a simple notification
result = send_notification(
title="Task Complete", message="Your task finished successfully!", timeout=5, urgency="normal"
)
# Send and automatically log notification
result = send_and_log_notification(
title="Status Update", message="System is running normally", log_dir="data"
)Dependencies: plyer>=2.1.0 (automatically installed)
Execute external Python scripts and CLI commands:
from helpers import execute_python_script, execute_cli_command, process_execution_results
# Execute a Python script
result = execute_python_script(
script_path="/path/to/script.py", args=["arg1", "arg2"], cwd="/working/directory", timeout=60
)
# Execute CLI commands
result = execute_cli_command(command="git status --porcelain", cwd="/repo/path")
# Process multiple execution results
summary = process_execution_results([result1, result2, result3])Dependencies: None (uses standard library)
Easy-to-use logging utilities with automatic task name detection:
from helpers import log_info, log_error, log_task_start, log_task_complete
@repeat(every(5).minutes)
def start():
log_task_start() # Automatically detects task name
log_info("Processing data...")
try:
# Your task logic here
result = process_data()
log_info(f"Processed {result['count']} items")
except Exception as e:
log_error("Failed to process data", exception=e)
raise
log_task_complete(duration=5.2)Key Features:
- ✅ Automatic task detection: No need to specify task names
- ✅ Structured logging: JSON-formatted logs for easy parsing
- ✅ Error tracking: Automatic exception capture with tracebacks
- ✅ Execution results: Log operation outcomes with details
- ✅ Multiple log files: Separate logs for tasks and errors
Available Functions:
log_info(),log_error(),log_warning(),log_debug()log_task_start(),log_task_complete(),log_task_error()log_execution_result(),log_structured_data()
Log Files:
logs/tasks.log: All task-specific logs with[task_name]prefixlogs/errors.log: Error logs from tasks and system issueslogs/scheduler.log: General scheduler operation logs
"""
---
title: "My Task"
description: "Description of what this task does"
dependencies:
- "requests>=2.25.0"
enabled: true
timeout: 300
---
"""
from task_scheduler.decorators import repeat, every
from datetime import datetime
@repeat(every(5).minutes) # Runs at :00, :05, :10, :15, etc.
def start():
"""Main task function - called by scheduler"""
print(f"Task executed at {datetime.now()}")
# Your task logic here# Every minute (high frequency monitoring)
@repeat(every(1).minutes)
# Every 5 minutes (regular checks)
@repeat(every(5).minutes)
# Every 30 minutes (periodic updates)
@repeat(every(30).minutes)
# Every hour (hourly processing)
@repeat(every(1).hours)
# Every day (daily maintenance)
@repeat(every(1).day)# Testing
python scripts/run_tests.py # Run all tests
python scripts/run_tests.py --fast # Skip slow tests
python scripts/run_tests.py --coverage # With coverage
# Database Management
python scripts/reset_schedules.py --dry-run # Preview schedule reset
python scripts/reset_schedules.py --future-only # Reset overdue tasks only
python scripts/check_overdue.py # Check current overdue tasks
# Process Management
python scripts/restart_scheduler.py --dry-run # Check scheduler status
python scripts/restart_scheduler.py # Restart scheduler
# Run Scheduler Directly
python main.py --config config/config.yaml # Run scheduler in foreground
sh scripts/run_scheduler.sh # Run scheduler in foreground via script
# Debugging
python scripts/debug_jobs.py # Debug job registration
tail -f logs/scheduler.log # Monitor scheduler logs
tail -f logs/tasks.log # Monitor task execution- Fork the repository
- Create a feature branch
- Add tests for new functionality
- Ensure all tests pass:
python scripts/run_tests.py - Run tests with coverage:
python scripts/run_tests.py --coverage - Submit a pull request
# Clone and setup
git clone https://github.com/kinged007/py-task-scheduler
cd py-task-scheduler
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
# Run tests to verify setup
pytest -v
# Run scheduler in development mode
python main.py --config config/config.yamlMIT License - see LICENSE file for details.