Process Scheduler Tools

process_scheduler_tools

from datetime import datetime, timedelta, timezone

scheduler_action(
    action="MODIFY_JOB",
    job_id="data_freshness_check",
    trigger={
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    }
)

{
    "type": "cron",
    "hour": 9,              # Hour (0-23)
    "minute": 0,            # Minute (0-59)
    "day_of_week": "mon",   # Day of week (mon, tue, wed, thu, fri, sat, sun or 0-6)
    "day": 1,               # Day of month (1-31)
    "month": "*",           # Month (1-12)
    "timezone": "UTC"       # Optional timezone
}

{
    "type": "interval",
    "seconds": 30,      # Optional: interval in seconds
    "minutes": 5,       # Optional: interval in minutes
    "hours": 2,         # Optional: interval in hours
    "days": 1           # Optional: interval in days
}

{
    "type": "date",
    "run_date": "2024-04-01T08:00:00"  # ISO format date/time
}

# Step 1: Check scheduler is running
status = scheduler_action(action="STATUS")
print(f"Scheduler status: {status}")

# Step 2: Create daily data refresh job (runs at 2 AM UTC)
scheduler_action(
    action="ADD_JOB",
    job_id="daily_customer_refresh",
    name="Daily Customer Data Refresh",
    what_to_do_prompt="""
Execute the following data refresh workflow:
1. Query the customer database for updates since last run
2. Transform data according to business rules
3. Load into analytics warehouse
4. Run data quality checks
5. If any issues found, send alert to #data-ops Slack channel
6. On success, update last_run timestamp
""",
    trigger={
        "type": "cron",
        "hour": 2,
        "minute": 0
    },
    coalesce=True,
    max_instances=1
)

# Step 3: Create frequent monitoring job (every 5 minutes)
from datetime import datetime, timedelta, timezone

scheduler_action(
    action="ADD_JOB",
    job_id="pipeline_health_check",
    name="Pipeline Health Monitor",
    what_to_do_prompt="""
Check health of all active data pipelines:
1. Query pipeline_status table for jobs running longer than expected
2. Check for any jobs in 'failed' state in the last hour
3. Verify data freshness for critical tables (< 2 hours old)
4. If any issues detected, send alert with details
5. Log check results to monitoring table
""",
    trigger={
        "type": "interval",
        "minutes": 5,
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    },
    max_instances=1
)

# Step 4: Create weekly report job (every Monday at 9 AM)
scheduler_action(
    action="ADD_JOB",
    job_id="weekly_executive_report",
    name="Weekly Executive Report",
    what_to_do_prompt="""
Generate weekly executive report:
1. Calculate KPIs for previous week (revenue, customers, growth metrics)
2. Generate trend charts for key metrics
3. Create summary document with insights
4. Save report to shared drive
5. Send email notification with report link to executive team
6. Post summary to #exec-updates Slack channel
""",
    trigger={
        "type": "cron",
        "hour": 9,
        "minute": 0,
        "day_of_week": "mon"
    }
)

# Step 5: Create end-of-month job (runs on the 1st at midnight)
scheduler_action(
    action="ADD_JOB",
    job_id="monthly_close_process",
    name="Monthly Financial Close",
    what_to_do_prompt="""
Execute month-end financial close process:
1. Run month-end reconciliation queries
2. Generate financial statements
3. Calculate month-over-month variance analysis
4. Archive previous month data
5. Initialize new month tracking tables
6. Send completion notification to finance team
""",
    trigger={
        "type": "cron",
        "hour": 0,
        "minute": 0,
        "day": 1
    },
    max_instances=1
)

# Step 6: List all scheduled jobs
jobs = scheduler_action(action="LIST_JOBS")
print(f"Total scheduled jobs: {len(jobs['jobs'])}")
for job in jobs['jobs']:
    print(f"- {job['name']}: next run at {job['next_run_time']}")

# Step 7: Run a job manually for testing
print("\\nTesting pipeline health check...")
run_result = scheduler_action(
    action="RUN_JOB",
    job_id="pipeline_health_check"
)
print(f"Manual run triggered: {run_result['run_id']}")

# Wait for execution to complete
import time
time.sleep(5)

# Step 8: Check execution results
run_details = scheduler_action(
    action="GET_RUN",
    run_id=run_result['run_id']
)
print(f"Run status: {run_details['status']}")
print(f"Output: {run_details['output']}")

# Step 9: View execution history for a job
history = scheduler_action(
    action="GET_HISTORY",
    job_id="pipeline_health_check",
    limit=5
)
print(f"\\nLast 5 executions of pipeline health check:")
for run in history['runs']:
    print(f"- {run['start_time']}: {run['status']} ({run['duration']}s)")

# Step 10: Modify job schedule (change to every 10 minutes instead of 5)
scheduler_action(
    action="MODIFY_JOB",
    job_id="pipeline_health_check",
    trigger={
        "type": "interval",
        "minutes": 10
    }
)
print("Updated health check to run every 10 minutes")

# Step 11: Pause job temporarily (e.g., during maintenance)
scheduler_action(
    action="PAUSE_JOB",
    job_id="daily_customer_refresh"
)
print("Paused daily refresh job for maintenance")

# Later: Resume the job
scheduler_action(
    action="RESUME_JOB",
    job_id="daily_customer_refresh"
)
print("Resumed daily refresh job")

# Step 12: Clean up old history (keep last 30 days)
scheduler_action(
    action="CLEAR_HISTORY",
    job_id="pipeline_health_check",
    older_than_days=30
)
print("Cleared history older than 30 days")

from datetime import datetime, timedelta, timezone

scheduler_action(
    action="ADD_JOB",
    job_id="smart_refresh",
    name="Conditional Data Refresh",
    what_to_do_prompt="""
Execute smart data refresh:
1. Check if source data has been updated since last run
2. If no new data, skip processing and log skip reason
3. If new data exists:
   a. Calculate approximate processing time based on data volume
   b. If > 10k new records, send notification that large batch is processing
   c. Execute refresh
   d. Validate results
   e. Send completion summary
""",
    trigger={
        "type": "interval",
        "minutes": 15,
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    }
)

scheduler_action(
    action="ADD_JOB",
    job_id="etl_pipeline",
    name="Complete ETL Pipeline",
    what_to_do_prompt="""
Execute complete ETL pipeline:

EXTRACT:
1. Pull data from production database (last 24 hours)
2. Extract data from API endpoints
3. Download files from S3 bucket

TRANSFORM:
4. Validate data quality (check for nulls, duplicates, schema)
5. Apply business logic transformations
6. Aggregate metrics by date and region
7. Enrich with reference data

LOAD:
8. Load to staging tables
9. Run merge to production tables
10. Update materialized views
11. Refresh BI tool cache

VALIDATE & NOTIFY:
12. Run post-load validation queries
13. Compare record counts before/after
14. Generate execution summary
15. Send success notification with metrics
16. If any step fails, rollback and alert immediately
""",
    trigger={
        "type": "cron",
        "hour": 3,
        "minute": 0
    },
    max_instances=1
)

# Define multiple similar jobs from configuration
data_sources = [
    {"name": "customers", "table": "dim_customers", "hour": 1},
    {"name": "orders", "table": "fact_orders", "hour": 2},
    {"name": "products", "table": "dim_products", "hour": 3},
    {"name": "inventory", "table": "fact_inventory", "hour": 4}
]

for source in data_sources:
    scheduler_action(
        action="ADD_JOB",
        job_id=f"refresh_{source['name']}",
        name=f"{source['name'].title()} Data Refresh",
        what_to_do_prompt=f"""
Refresh {source['name']} data:
1. Query source system for {source['name']} updates
2. Stage in temp table
3. Merge to {source['table']}
4. Log row counts and execution time
5. Update metadata table with refresh timestamp
""",
        trigger={
            "type": "cron",
            "hour": source['hour'],
            "minute": 0
        }
    )
    print(f"Created refresh job for {source['name']}")

from datetime import datetime, timedelta, timezone

scheduler_action(
    action="ADD_JOB",
    job_id="error_recovery",
    name="Automatic Error Recovery",
    what_to_do_prompt="""
Check for and recover from pipeline errors:
1. Query error_log table for failures in last hour
2. For each failed job:
   a. Check if error is transient (connection timeout, lock, etc.)
   b. If transient, attempt retry with exponential backoff
   c. If persistent, escalate to on-call engineer
   d. Log recovery attempt and outcome
3. Update error_log with recovery status
4. If multiple failures of same type, create incident ticket
""",
    trigger={
        "type": "interval",
        "minutes": 15,
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    }
)

scheduler_action(
    action="ADD_JOB",
    job_id="nightly_cleanup",
    name="Nightly System Cleanup",
    what_to_do_prompt="""
Perform system maintenance and cleanup:
1. Archive logs older than 90 days
2. Delete temporary tables created > 7 days ago
3. Vacuum and analyze frequently updated tables
4. Clear cached query results older than 24 hours
5. Remove orphaned files from staging areas
6. Compact transaction log files
7. Generate cleanup summary report
8. Log space reclaimed and performance improvements
""",
    trigger={
        "type": "cron",
        "hour": 1,
        "minute": 30
    }
)

def monitor_scheduled_jobs():
    """Monitor health of all scheduled jobs"""
    # Get all jobs
    jobs = scheduler_action(action="LIST_JOBS")
    
    issues = []
    
    for job in jobs['jobs']:
        # Get recent history
        history = scheduler_action(
            action="GET_HISTORY",
            job_id=job['job_id'],
            limit=5
        )
        
        # Check for consecutive failures
        recent_runs = history['runs']
        if len(recent_runs) >= 3:
            if all(run['status'] == 'failed' for run in recent_runs[:3]):
                issues.append({
                    'job_id': job['job_id'],
                    'issue': 'Three consecutive failures',
                    'severity': 'high'
                })
        
        # Check for long-running jobs
        for run in recent_runs:
            if run['status'] == 'running' and run['duration'] > 3600:
                issues.append({
                    'job_id': job['job_id'],
                    'issue': f'Job running for {run["duration"]}s',
                    'severity': 'medium'
                })
        
        # Check for stale jobs (not run in expected timeframe)
        if job['last_run_time']:
            from datetime import datetime
            last_run = datetime.fromisoformat(job['last_run_time'])
            now = datetime.now()
            hours_since = (now - last_run).total_seconds() / 3600
            
            # For hourly jobs, alert if no run in 3 hours
            if 'interval' in job['trigger'] and hours_since > 3:
                issues.append({
                    'job_id': job['job_id'],
                    'issue': f'No execution in {hours_since:.1f} hours',
                    'severity': 'medium'
                })
    
    return issues

# Run monitoring
issues = monitor_scheduled_jobs()
if issues:
    print(f"Found {len(issues)} issues:")
    for issue in issues:
        print(f"⚠️ {issue['job_id']}: {issue['issue']} [{issue['severity']}]")
else:
    print("✅ All scheduled jobs healthy")

# Old cron entry
0 9 * * * /path/to/script.sh

# New scheduled job
scheduler_action(
    action="ADD_JOB",
    job_id="daily_task",
    what_to_do_prompt="Execute the daily task script",
    trigger={"type": "cron", "hour": 9, "minute": 0}
)