Scheduled & Event-Based Automation: When and How to Trigger

Automation Triggers and Scheduling covers the mechanisms that initiate automated processes based on time intervals, events, or conditions—essential for creating responsive and timely automation systems.

Table of Contents

• Scheduled Automation
• Event-Based Automation

Scheduled Automation

Scheduled automation involves executing automated tasks at predetermined times or intervals using time-based triggers, enabling regular maintenance, reporting, and batch processing without manual intervention.

Why it is useful: Scheduled automation ensures timely execution of routine tasks, enables off-peak processing to minimize system impact, provides predictable automation windows, and supports regular business cycles like daily reports or monthly billing.

Common Scheduled Automation Tasks

• Daily data backups and system maintenance
• Regular report generation and distribution
• Periodic data synchronization between systems
• Routine cleanup and maintenance operations
• Batch processing of transactions or records
• Scheduled email newsletters and communications
• Regular security scans and compliance checks
• Automated system updates and patch deployments

Scheduling Mechanisms and Tools

• Cron (Unix/Linux): Time-based job scheduler using cron expressions
• Windows Task Scheduler: GUI and command-line task scheduling for Windows
• Cloud Schedulers: AWS EventBridge, Google Cloud Scheduler, Azure Timer Triggers
• Python Libraries: schedule, APScheduler, Celery Beat
• Workflow Engines: Apache Airflow, Luigi, Prefect (with built-in scheduling)
• Application-Specific: Built-in scheduling in automation platforms (Zapier, Make)

Cron Expression Basics

• Format: * * * * * command-to-execute
• Fields: Minute (0-59), Hour (0-23), Day of Month (1-31), Month (1-12), Day of Week (0-6, Sunday=0)
• Special Characters:
  • *: Every value
  • ,: List of values (e.g., 1,3,5)
  • -: Range of values (e.g., 9-17)
  • /: Step values (e.g., */15 every 15 minutes)

Example: Scheduled Backup System with Python

import schedule
import time
import shutil
import os
from datetime import datetime
import logging

class ScheduledBackup:
    def __init__(self, source_dir, backup_dir, retention_days=30):
        self.source_dir = source_dir
        self.backup_dir = backup_dir
        self.retention_days = retention_days
        self.setup_logging()
    
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    
    def create_backup(self):
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_path = os.path.join(self.backup_dir, f"backup_{timestamp}")
        
        try:
            shutil.copytree(self.source_dir, backup_path)
            self.logger.info(f"Backup created: {backup_path}")
            self.cleanup_old_backups()
        except Exception as e:
            self.logger.error(f"Backup failed: {e}")
    
    def cleanup_old_backups(self):
        cutoff = time.time() - (self.retention_days * 86400)
        for folder in os.listdir(self.backup_dir):
            folder_path = os.path.join(self.backup_dir, folder)
            if os.path.getmtime(folder_path) < cutoff:
                shutil.rmtree(folder_path)
                self.logger.info(f"Removed old backup: {folder}")
    
    def run_scheduler(self):
        # Daily at 2 AM
        schedule.every().day.at("02:00").do(self.create_backup)
        # Weekly on Sunday at 3 AM
        schedule.every().sunday.at("03:00").do(self.create_backup)
        
        self.logger.info("Scheduler started")
        while True:
            schedule.run_pending()
            time.sleep(60)

# Usage
# backup = ScheduledBackup("/data", "/backups")
# backup.run_scheduler()

Cron Alternative (crontab)

# Edit with: crontab -e

# Daily backup at 2:00 AM
0 2 * * * /usr/bin/python3 /scripts/daily_backup.py >> /var/log/backup.log 2>&1

# Weekly full backup Sunday 3:00 AM
0 3 * * 0 /usr/bin/python3 /scripts/weekly_backup.py >> /var/log/backup.log 2>&1

# Cleanup old backups daily at 4:00 AM
0 4 * * * /usr/bin/python3 /scripts/cleanup_backups.py >> /var/log/backup.log 2>&1

Best Practices

• Schedule resource-intensive tasks during off-peak hours
• Implement proper logging and monitoring for scheduled tasks
• Include verification steps to ensure task completion
• Handle failures gracefully with alerts and retry mechanisms
• Test schedules in development environments before production
• Document all scheduled tasks and their purposes
• Regularly review and optimize schedules based on system load

Event-Based Automation

Event-based automation involves executing automated processes in response to specific occurrences or changes in systems, applications, or environments—enabling real-time responsiveness and reactive workflows.

Why it is useful: Event-based automation provides immediate response to important changes, reduces polling overhead, enables real-time workflows, and creates more responsive and efficient automation systems compared to pure scheduled approaches.

Common Event-Based Automation Tasks

• Real-time notifications for system errors, security alerts, or performance issues
• Automated responses to customer actions (form submissions, purchases, support tickets)
• Data synchronization when source data changes
• Workflow triggers based on document approvals or form completions
• Inventory updates when stock levels change
• Social media monitoring and response automation
• IoT device responses to sensor readings or device status changes
• Git repository webhook responses to code commits or pull requests

Event Sources and Mechanisms

• Webhooks: HTTP callbacks from applications when events occur
• Message Queues: RabbitMQ, Apache Kafka, Amazon SQS, Azure Service Bus
• File System Watchers: Monitoring directories for file changes
• Database Triggers: SQL triggers or change data capture (CDC)
• Application Events: Built-in event systems in frameworks and platforms
• Cloud Events: AWS SNS, Google Pub/Sub, Azure Event Grid

Example: File System Watcher for Document Processing

import time
import os
import shutil
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import logging

class DocumentHandler(FileSystemEventHandler):
    def __init__(self, watch_dir, processed_dir):
        self.watch_dir = watch_dir
        self.processed_dir = processed_dir
        os.makedirs(processed_dir, exist_ok=True)
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def on_created(self, event):
        if not event.is_directory:
            self.process_file(event.src_path)
    
    def on_moved(self, event):
        if not event.is_directory:
            self.process_file(event.dest_path)
    
    def process_file(self, file_path):
        filename = os.path.basename(file_path)
        self.logger.info(f"Processing: {filename}")
        
        time.sleep(1)  # Wait for file write to complete
        
        try:
            # Move to processed folder with timestamp
            name, ext = os.path.splitext(filename)
            timestamp = time.strftime("%Y%m%d_%H%M%S")
            new_name = f"{name}_{timestamp}{ext}"
            dest = os.path.join(self.processed_dir, new_name)
            shutil.move(file_path, dest)
            self.logger.info(f"Moved {filename} to processed")
        except Exception as e:
            self.logger.error(f"Error processing {filename}: {e}")

def start_watcher(watch_dir, processed_dir):
    handler = DocumentHandler(watch_dir, processed_dir)
    observer = Observer()
    observer.schedule(handler, watch_dir, recursive=False)
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

# Usage
# start_watcher("./incoming", "./processed")

Example: Webhook Handler for GitHub Events

from flask import Flask, request, jsonify
import hmac
import hashlib
import os

app = Flask(__name__)
WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET")

@app.route("/webhook/github", methods=["POST"])
def github_webhook():
    # Verify signature
    signature = request.headers.get("X-Hub-Signature-256", "")
    payload = request.get_data()
    
    if not verify_signature(payload, signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    event_type = request.headers.get("X-GitHub-Event", "")
    data = request.json
    
    if event_type == "push":
        handle_push_event(data)
    elif event_type == "pull_request":
        handle_pr_event(data)
    
    return jsonify({"status": "ok"})

def verify_signature(payload, signature):
    if not WEBHOOK_SECRET:
        return True  # Skip in dev
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

def handle_push_event(data):
    repo = data["repository"]["full_name"]
    commits = len(data["commits"])
    print(f"Push to {repo}: {commits} commits")
    # Trigger CI/CD, deploy, notify team, etc.

def handle_pr_event(data):
    action = data["action"]
    pr = data["pull_request"]["number"]
    print(f"PR #{pr} {action}")
    # Run tests, notify reviewers, etc.

# Run: python app.py

Best Practices

• Implement idempotency to handle duplicate events safely
• Use dead letter queues for failed event processing
• Implement proper error handling and logging
• Consider event ordering and consistency requirements
• Monitor event processing lag and throughput
• Secure event endpoints and validate incoming events
• Test event handlers with various event types and payloads
• Document event schemas and processing logic