smoothschedule/smoothschedule/SCHEDULER_PLUGIN_SYSTEM.md

Resource-Free Scheduler & Plugin System

Overview

The Resource-Free Scheduler is a comprehensive system for running automated tasks on schedules without requiring resource allocation. It's separate from the customer-facing Event/Appointment system and designed for internal automation tasks.

Key Features

• Plugin-Based Architecture: Extensible system for creating custom automated tasks
• Multiple Schedule Types: Cron expressions, fixed intervals, and one-time executions
• Celery Integration: Asynchronous task execution with retry logic
• Execution Logging: Complete audit trail of all task executions
• Multi-Tenant Aware: Works seamlessly with django-tenants
• Built-in Plugins: Common tasks ready to use out of the box

Architecture

┌─────────────────┐
│ ScheduledTask   │  (Model: Configuration for recurring tasks)
│ - plugin_name   │
│ - schedule_type │
│ - config        │
└────────┬────────┘
         │
         ├─> Plugin Registry ──> BasePlugin ──> Custom Plugins
         │                                          │
         │                                          ├─> SendEmailPlugin
         │                                          ├─> DailyReportPlugin
         │                                          ├─> CleanupPlugin
         │                                          └─> WebhookPlugin
         │
         └─> Celery Tasks ──> execute_scheduled_task()
                                     │
                                     └─> TaskExecutionLog (Model: Execution history)

Core Components

1. Models (schedule/models.py)

ScheduledTask

Stores configuration for scheduled tasks.

Fields:

• name - Human-readable task name
• description - What the task does
• plugin_name - Which plugin to execute
• plugin_config - JSON configuration for the plugin
• schedule_type - CRON, INTERVAL, or ONE_TIME
• cron_expression - Cron syntax (e.g., "0 0 * * *")
• interval_minutes - Run every N minutes
• run_at - Specific datetime for one-time tasks
• status - ACTIVE, PAUSED, or DISABLED
• next_run_at - Calculated next execution time
• last_run_at - When last executed
• last_run_status - success/failed
• last_run_result - JSON result from last execution

TaskExecutionLog

Audit log of all task executions.

Fields:

• scheduled_task - Reference to the task
• started_at - Execution start time
• completed_at - Execution end time
• status - SUCCESS, FAILED, or SKIPPED
• result - JSON result from plugin
• error_message - Error details if failed
• execution_time_ms - Duration in milliseconds

2. Plugin System (schedule/plugins.py)

BasePlugin

Abstract base class for all plugins.

Required Attributes:

• name - Unique identifier (snake_case)
• display_name - Human-readable name
• description - What the plugin does
• category - For organization (e.g., "communication", "reporting")

Required Methods:

• execute(context) - Main task logic

Optional Methods:

• validate_config() - Validate plugin configuration
• can_execute(context) - Pre-execution checks
• on_success(result) - Post-success callback
• on_failure(error) - Error handling callback

Plugin Registry

Global registry for managing plugins.

Methods:

• register(plugin_class) - Register a plugin
• get(plugin_name) - Get plugin class by name
• get_instance(plugin_name, config) - Create plugin instance
• list_all() - List all plugins with metadata
• list_by_category() - Group plugins by category

Template Variables (schedule/template_parser.py)

Template variables allow plugins to define configurable fields that users can fill in via the UI.

Variable Format:

{{PROMPT:variable_name|description}}
{{PROMPT:variable_name|description|default_value}}
{{PROMPT:variable_name|description|default_value|type}}
{{PROMPT:variable_name|description||type}}  (no default, explicit type)

Supported Types:

Type	Description	UI Component
text	Single-line text input	Text input
textarea	Multi-line text input	Textarea
email	Email address	Email input with validation
number	Numeric value	Number input
url	URL/webhook endpoint	URL input with validation
email_template	Email template selector	Dropdown of email templates

Type Inference: If no explicit type is provided, the parser infers type from the variable name:

• Names containing email → email type
• Names containing count, days, hours, amount → number type
• Names containing url, webhook, endpoint → url type
• Names containing message, body, content → textarea type
• Default → text type

Example - Email Template Variable:

# In your plugin script, reference an email template:
template_id = {{PROMPT:confirmation_template|Email template for confirmations||email_template}}

# The UI will show a dropdown of all available email templates
# The user selects a template, and the template ID is stored in the config

Using Email Templates in Plugins:

from schedule.models import EmailTemplate

class MyNotificationPlugin(BasePlugin):
    def execute(self, context):
        template_id = self.config.get('confirmation_template')

        if template_id:
            template = EmailTemplate.objects.get(id=template_id)
            subject, html, text = template.render({
                'BUSINESS_NAME': context['business'].name,
                'CUSTOMER_NAME': customer.name,
                # ... other variables
            })

            # Send the email using the rendered template
            send_email(recipient, subject, html, text)

Insertion Codes (for use within templates):

These codes are replaced at runtime with actual values:

• {{BUSINESS_NAME}} - Business name
• {{BUSINESS_EMAIL}} - Business contact email
• {{BUSINESS_PHONE}} - Business phone number
• {{CUSTOMER_NAME}} - Customer's name
• {{CUSTOMER_EMAIL}} - Customer's email
• {{APPOINTMENT_TIME}} - Appointment date/time
• {{APPOINTMENT_DATE}} - Appointment date only
• {{APPOINTMENT_SERVICE}} - Service name
• {{TODAY}} - Today's date
• {{NOW}} - Current date and time

3. Celery Tasks (schedule/tasks.py)

execute_scheduled_task(scheduled_task_id)

Main task executor. Runs the plugin, logs results, updates next run time.

Features:

• Automatic retry with exponential backoff
• Multi-tenant context preservation
• Pre-execution validation
• Comprehensive error handling
• Automatic next-run calculation

check_and_schedule_tasks()

Background task that finds due tasks and queues them for execution.

Built-in Plugins

1. SendEmailPlugin

Send custom emails to recipients.

Config:

{
  "recipients": ["user@example.com"],
  "subject": "Subject line",
  "message": "Email body",
  "from_email": "sender@example.com"  // optional
}

2. CleanupOldEventsPlugin

Delete old completed/canceled events.

Config:

{
  "days_old": 90,
  "statuses": ["COMPLETED", "CANCELED"],
  "dry_run": false
}

3. DailyReportPlugin

Send daily business summary reports.

Config:

{
  "recipients": ["manager@example.com"],
  "include_upcoming": true,
  "include_completed": true
}

4. AppointmentReminderPlugin

Send appointment reminders to customers.

Config:

{
  "hours_before": 24,
  "method": "email",  // or "sms", "both"
  "message_template": "Optional custom template"
}

5. BackupDatabasePlugin

Create database backups.

Config:

{
  "backup_location": "/path/to/backups",
  "compress": true
}

6. WebhookPlugin

Call external webhook URLs.

Config:

{
  "url": "https://example.com/webhook",
  "method": "POST",
  "headers": {"Authorization": "Bearer token"},
  "payload": {"key": "value"}
}

API Endpoints

Scheduled Tasks

List tasks:

GET /api/scheduled-tasks/

Create task:

POST /api/scheduled-tasks/
{
  "name": "Daily Cleanup",
  "description": "Clean up old events",
  "plugin_name": "cleanup_old_events",
  "plugin_config": {"days_old": 90},
  "schedule_type": "CRON",
  "cron_expression": "0 0 * * *",
  "status": "ACTIVE"
}

Update task:

PATCH /api/scheduled-tasks/{id}/

Pause task:

POST /api/scheduled-tasks/{id}/pause/

Resume task:

POST /api/scheduled-tasks/{id}/resume/

Manually execute:

POST /api/scheduled-tasks/{id}/execute/

Get execution logs:

GET /api/scheduled-tasks/{id}/logs/?limit=20&offset=0

Plugins

List all plugins:

GET /api/plugins/

List by category:

GET /api/plugins/by_category/

Get plugin details:

GET /api/plugins/{plugin_name}/

Execution Logs

List all logs:

GET /api/task-logs/

Filter by task:

GET /api/task-logs/?task_id=1

Filter by status:

GET /api/task-logs/?status=FAILED

Creating Custom Plugins

Step 1: Create Plugin Class

# myapp/plugins.py
from schedule.plugins import BasePlugin, register_plugin

@register_plugin
class MyCustomPlugin(BasePlugin):
    name = "my_custom_plugin"
    display_name = "My Custom Plugin"
    description = "Does something awesome"
    category = "automation"

    config_schema = {
        'api_key': {
            'type': 'string',
            'required': True,
            'description': 'API key for external service',
        },
        'threshold': {
            'type': 'integer',
            'required': False,
            'default': 100,
            'description': 'Processing threshold',
        },
        'notification_template': {
            'type': 'email_template',
            'required': False,
            'description': 'Email template to use for notifications',
        },
    }

    def execute(self, context):
        """
        context contains:
        - business: Current tenant
        - scheduled_task: ScheduledTask instance
        - execution_time: When execution started
        - user: User who created the task
        """
        api_key = self.config.get('api_key')
        threshold = self.config.get('threshold', 100)

        # Do your task logic here
        result = perform_some_operation(api_key, threshold)

        return {
            'success': True,
            'message': 'Operation completed',
            'data': {
                'items_processed': result.count,
            }
        }

    def can_execute(self, context):
        """Optional: Add pre-execution checks"""
        if context['business'].is_suspended:
            return False, "Business is suspended"
        return True, None

    def on_failure(self, error):
        """Optional: Handle failures"""
        # Send alert, log to external service, etc.
        pass

Step 2: Import Plugin on Startup

Add to your app's apps.py:

class MyAppConfig(AppConfig):
    def ready(self):
        from . import plugins  # Import to register

Step 3: Use via API

curl -X POST http://lvh.me:8000/api/scheduled-tasks/ \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Custom Automation",
    "plugin_name": "my_custom_plugin",
    "plugin_config": {
      "api_key": "secret123",
      "threshold": 500
    },
    "schedule_type": "INTERVAL",
    "interval_minutes": 60,
    "status": "ACTIVE"
  }'

Schedule Types

Cron Expression

{
  "schedule_type": "CRON",
  "cron_expression": "0 0 * * *"  # Daily at midnight
}

# Examples:
# "*/5 * * * *"   - Every 5 minutes
# "0 */2 * * *"   - Every 2 hours
# "0 9 * * 1-5"   - Weekdays at 9am
# "0 0 1 * *"     - First day of month

Fixed Interval

{
  "schedule_type": "INTERVAL",
  "interval_minutes": 30  # Every 30 minutes
}

One-Time

{
  "schedule_type": "ONE_TIME",
  "run_at": "2025-12-01T10:00:00Z"
}

Execution Context

Every plugin receives a context dictionary:

{
    'business': <Business instance>,           # Current tenant
    'scheduled_task': <ScheduledTask instance>,
    'execution_time': <datetime>,              # Execution timestamp
    'user': <User instance or None>,           # Task creator
}

Error Handling

Tasks automatically retry on failure with exponential backoff:

1. First retry: 60 seconds
2. Second retry: 120 seconds
3. Third retry: 240 seconds

After 3 failures, the task is marked as failed and won't retry until next scheduled run.

Multi-Tenant Behavior

The scheduler respects django-tenants:

• Tasks execute in the context of their tenant's schema
• Each tenant has separate scheduled tasks
• Execution logs are tenant-isolated
• Plugins can access context['business'] for tenant info

Monitoring & Debugging

View Recent Logs

curl http://lvh.me:8000/api/task-logs/?limit=10

Check Failed Tasks

curl http://lvh.me:8000/api/task-logs/?status=FAILED

View Task Details

curl http://lvh.me:8000/api/scheduled-tasks/1/

Manual Test Execution

curl -X POST http://lvh.me:8000/api/scheduled-tasks/1/execute/

Best Practices

1. Plugin Design

• Keep plugins focused on one task
• Make config schema explicit
• Return structured results
• Handle errors gracefully
• Add logging for debugging

2. Configuration

• Use environment variables for secrets (not plugin_config)
• Validate config in validate_config()
• Provide sensible defaults
• Document config schema

3. Scheduling

• Use cron for predictable times (e.g., daily reports)
• Use intervals for continuous processing
• Avoid very short intervals (< 5 minutes)
• Consider business hours for customer-facing tasks

4. Error Handling

• Use can_execute() for pre-flight checks
• Return detailed error messages
• Implement on_failure() for alerts
• Don't silence exceptions in execute()

5. Performance

• Keep executions fast (< 30 seconds ideal)
• Use pagination for large datasets
• Offload heavy work to dedicated queues
• Monitor execution times in logs

Testing

Test Plugin Directly

from schedule.plugins import registry

plugin = registry.get_instance('my_plugin', config={'key': 'value'})
result = plugin.execute({
    'business': business,
    'scheduled_task': None,
    'execution_time': timezone.now(),
    'user': None,
})
assert result['success'] == True

Test via API

import requests

# Create task
response = requests.post('http://lvh.me:8000/api/scheduled-tasks/', json={
    'name': 'Test Task',
    'plugin_name': 'send_email',
    'plugin_config': {'recipients': ['test@example.com']},
    'schedule_type': 'ONE_TIME',
    'run_at': '2025-12-01T10:00:00Z',
})
task_id = response.json()['id']

# Execute immediately
requests.post(f'http://lvh.me:8000/api/scheduled-tasks/{task_id}/execute/')

# Check logs
logs = requests.get(f'http://lvh.me:8000/api/scheduled-tasks/{task_id}/logs/')

Troubleshooting

Plugins Not Registered

Check that apps.py imports the plugin module in ready().

Tasks Not Executing

1. Check task status is ACTIVE
2. Verify next_run_at is in the future
3. Check Celery worker is running
4. Look for errors in Django logs

Execution Logs Empty

Check that execute_scheduled_task Celery task is configured and workers are running.

Cron Expression Not Working

Use a cron validator. Common issues:

• Syntax errors
• Timezone mismatches (server runs in UTC)

Future Enhancements

Potential additions:

• Web UI for managing tasks
• Plugin marketplace
• Task dependencies/chaining
• Rate limiting
• Notifications on failure
• Dashboard with metrics
• Plugin versioning
• A/B testing for plugins