task-automator

name	task-automator
description	Task automation design expertise covering automation opportunity identification, script design patterns, robust error handling, structured logging, scheduling strategies, failure notification, idempotency principles, dry-run mode implementation, configuration management, and documentation standards for maintainable automation. Use when the user asks about task automator, task automator best practices, or needs guidance on task automator implementation. Do NOT use when the user needs a different specialized skill or is asking about an unrelated technology domain.
license	Apache-2.0
metadata	{"author":"foundry-skills","version":"1.0.0","tags":"automation shell-scripting guide","category":"software-engineering","subcategory":"developer-tools","depends":"","disclaimer":"none","difficulty":"intermediate"}

Task Automator

Core Philosophy

Automation exists to eliminate repetitive manual work, reduce human error, and ensure consistency. A well-designed automation script is idempotent, observable, recoverable, and documented. If a task is performed more than twice and follows a predictable pattern, it is a candidate for automation.

Automation Opportunity Identification

Decision Framework

Is this task:
  [x] Performed more than 2-3 times per week?
  [x] Following a predictable, repeatable pattern?
  [x] Taking more than 5 minutes manually?
  [x] Error-prone when done manually?
  [x] Blocking other work while waiting for completion?

If 3+ boxes checked -> Automate it.

Prioritization formula:
  Value = (Time_saved_per_execution * Frequency) / Development_effort

High value automation targets:
  - Data backups and rotation
  - Environment setup and teardown
  - Report generation and distribution
  - Deployment and release processes
  - Log rotation and cleanup
  - Certificate renewal
  - Health checks and monitoring
  - Data synchronization between systems
  - Onboarding/offboarding user accounts
  - Dependency updates and security patches

Script Design Patterns

Template: Robust Automation Script

#!/usr/bin/env python3
"""
Daily Report Generator

Generates and distributes daily sales reports.
Designed to be run via cron at 06:00 UTC daily.

Usage:
    python daily_report.py
    python daily_report.py --dry-run
    python daily_report.py --date 2025-03-15
    python daily_report.py --config /path/to/config.yaml
"""

import argparse
import logging
import sys
import time
from datetime import date, timedelta
from pathlib import Path
from dataclasses import dataclass, field

# ──────────────────────────────────────────────
# Configuration
# ──────────────────────────────────────────────

@dataclass
class Config:
    """Configuration with sensible defaults and supersede support."""
    report_date: date = field(default_factory=lambda: date.today() - timedelta(days=1))
    output_dir: Path = Path("/var/reports/daily")
    recipients: list[str] = field(default_factory=lambda: ["[e-mail]"])
    database_url: str = ""
    smtp_host: str = "smtp.example.com"
    dry_run: bool = False
    verbose: bool = False

    @classmethod
    def from_args(cls, args: argparse.Namespace) -> 'Config':
        config = cls()
        if args.date:
            config.report_date = date.fromisoformat(args.date)
         args.dry_run:
            config.dry_run = 
         args.verbose:
            config.verbose = 
         args.output_dir:
            config.output_dir = Path(args.output_dir)
         config





 () -> logging.Logger:
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG  verbose  logging.INFO)

    
    console = logging.StreamHandler(sys.stdout)
    console.setFormatter(logging.Formatter(
        ,
        datefmt=
    ))
    logger.addHandler(console)

    
    file_handler = logging.FileHandler()
    file_handler.setFormatter(logging.Formatter(
        
    ))
    logger.addHandler(file_handler)

     logger





 :
     ():
        .config = config
        .logger = logger
        .stats = {: , : , : }

     () -> :
        
        start_time = time.time()
        .logger.info()

         .config.dry_run:
            .logger.info()

        :
            
            ._validate()

            
            data = ._fetch_data()

            
            report_path = ._generate_report(data)

            
            ._distribute(report_path)

            elapsed = time.time() - start_time
            .logger.info(
                
                
            )
             

         Exception  e:
            .stats[] += 
            elapsed = time.time() - start_time
            .logger.error(, exc_info=)
            ._notify_failure((e))
             

     ():
        
        .config.output_dir.mkdir(parents=, exist_ok=)
        
        .logger.debug()

     () -> :
        
        .logger.info()
        
        .stats[] += 
        .stats[] += 
         {: , : }

     () -> Path:
        
        report_path = .config.output_dir / 
         .config.dry_run:
            .logger.info()
             report_path
        
        .logger.info()
         report_path

     ():
        
         recipient  .config.recipients:
             .config.dry_run:
                .logger.info()
            :
                
                .logger.info()

     ():
        
        .logger.info()
        





 () -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=)
    parser.add_argument(, =)
    parser.add_argument(, action=, =)
    parser.add_argument(, , action=, =)
    parser.add_argument(, =)
    parser.add_argument(, =)
     parser.parse_args()

 ():
    args = parse_args()
    config = Config.from_args(args)
    logger = setup_logging(verbose=config.verbose)

    generator = ReportGenerator(config, logger)
    success = generator.run()

    sys.exit(  success  )

 __name__ == :
    main()

Error Handling

Retry with Exponential Backoff

import time
import functools

def retry(max_attempts=3, base_delay=1.0, max_delay=60.0, exceptions=(Exception,)):
    """Decorator for retry with exponential backoff."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        raise
                    delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
                    logging.warning(
                        f"{func.__name__} failed (attempt {attempt}/{max_attempts}): {e}. "
                        f"Retrying in {delay:.1f}s..."
                    )
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, base_delay=2.0, exceptions=(ConnectionError, TimeoutError))
 ():
    response = httpx.get(url, timeout=)
    response.raise_for_status()
     response.json()

Graceful Shutdown

import signal

class GracefulShutdown:
    def __init__(self):
        self.should_stop = False
        signal.signal(signal.SIGTERM, self._handler)
        signal.signal(signal.SIGINT, self._handler)

    def _handler(self, signum, frame):
        logging.info(f"Received signal {signum}, shutting down gracefully...")
        self.should_stop = True

# Usage in a loop
shutdown = GracefulShutdown()

for item in work_queue:
    if shutdown.should_stop:
        logging.info("Stopping work loop due to shutdown signal")
        break
    process(item)

Idempotency

class IdempotentProcessor:
    """Process items exactly once using a state file."""

    def __init__(self, state_file: Path):
        self.state_file = state_file
        self.processed = self._load_state()

    def _load_state(self) -> set:
        if self.state_file.exists():
            return set(self.state_file.read_text().splitlines())
        return set()

    def _save_state(self):
        self.state_file.write_text('\n'.join(sorted(self.processed)))

    def should_process(self, item_id: str) -> bool:
        return item_id not in self.processed

    def mark_processed(self, item_id: str):
        self.processed.add(item_id)
        self._save_state()

# Usage
processor = IdempotentProcessor(Path("/var/state/migration.state"))

for record in records:
    if  processor.should_process(record.):
        logger.debug()
        
    migrate_record(record)
    processor.mark_processed(record.)

Dry-Run Mode

class FileManager:
    def __init__(self, dry_run: bool = False):
        self.dry_run = dry_run
        self.logger = logging.getLogger(__name__)

    def move_file(self, source: Path, destination: Path):
        if self.dry_run:
            self.logger.info(f"[DRY RUN] Would move: {source} -> {destination}")
            return
        destination.parent.mkdir(parents=True, exist_ok=True)
        source.rename(destination)
        self.logger.info(f"Moved: {source} -> {destination}")

    def delete_file(self, path: Path):
        if self.dry_run:
            self.logger.info(f"[DRY RUN] Would delete: {path}")
            return
        path.unlink()
        self.logger.info(f"Deleted: {path}")

Notification on Failure

import smtplib
from email.mime.text import MIMEText

def send_alert(subject: str, body: str, recipients: list[str]):
    msg = MIMEText(body)
    msg['Subject'] = f"[AUTOMATION ALERT] {subject}"
    msg['From'] = '[e-mail]'
    msg['To'] = ', '.join(recipients)

    with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login('[e-mail]', 'password')
        server.send_message(msg)

# Slack notification
import httpx

def send_slack_alert(webhook_url: str, message: str, severity: str = "warning"):
    color = {"info": "#36a64f", "warning": "#ff9900", "error": "#ff0000"}[severity]
    httpx.post(webhook_url, json={
        "attachments": [{
            "color": color,
            "title": "Automation Alert",
            "text": message,
            "ts": int(time.time())
        }]
    })

Configuration Management

# config.yaml
report:
  output_dir: [system-path]
  format: pdf
  retention_days: 90

database:
  url: postgresql://user:pass@host/db
  pool_size: 5

notifications:
  email:
    recipients: ["[e-mail]", "[e-mail]"]
    smtp_host: smtp.example.com
  slack:
    webhook_url: [reference URL]
    channel: "#automation-alerts"

scheduling:
  timezone: UTC
  run_time: "06:00"

import yaml
from pathlib import Path

def load_config(config_path: Path | None = None) -> dict:
    """Load config from file, with env var supersedes."""
    # Default config path
    if config_path is None:
        config_path = Path(__file__).parent / "config.yaml"

    with open(config_path) as f:
        config = yaml.safe_load(f)

    # Environment variable supersedes
    import os
    if db_url := environment-variables.get("DATABASE_URL"):
        config["database"]["url"] = db_url
    if slack_url := environment-variables.get("SLACK_WEBHOOK_URL"):
        config["notifications"]["slack"]["webhook_url"] = slack_url

    return config

Documentation Standards

"""
Script: daily_report.py
Purpose: Generate and distribute daily sales reports
Author: Data Engineering Team
Created: 2025-01-15
Last Modified: 2025-03-15

Dependencies:
  - Python 3.11+
  - httpx, jinja2, weasyprint, pyyaml
  - PostgreSQL access (read-only)

Environment Variables:
  DATABASE_URL: PostgreSQL connection string (required)
  SLACK_WEBHOOK_URL: Slack notification webhook (optional)
  SMTP_PASSWORD: Email server password (required for email delivery)

Scheduling:
  Runs daily at 06:00 UTC via cron:
  0 6 * * * [system-path] [system-path] >> output_file 2>&1

Recovery:
  If the script fails, it can be safely re-run with --date YYYY-MM-DD.
  It is idempotent: re-running overwrites the previous report for that date.

Monitoring:
  - Check [system-path] for errors
  - Slack alerts sent to #automation-alerts on failure
  - Grafana dashboard: Automation > Daily Reports
"""

Best Practices

1. Always implement dry-run mode: Test changes safely before executing
2. Make scripts idempotent: Safe to run multiple times with same result
3. Use structured logging: Include timestamps, levels, and context
4. Fail loudly: Send notifications on failure, exit with non-zero code
5. Use configuration files: Avoid hardcoded values
6. Handle signals gracefully: Respond to SIGTERM for clean shutdown
7. Add timeouts: Every external call should have a timeout
8. Document everything: Purpose, dependencies, scheduling, recovery procedures
9. Version control your scripts: Track changes like any other code
10. Test automation scripts: Unit test business logic, integration test the full flow

When to Use

Use this skill when:

• Designing or implementing task automator solutions
• Reviewing or improving existing task automator approaches
• Making architectural or implementation decisions about task automator
• Learning task automator patterns and best practices
• Troubleshooting task automator-related issues

Do NOT use this skill when:

• The question is about a fundamentally different technology domain
• A more specific sibling skill covers the exact topic needed
• The user needs a complete hands-on tutorial rather than expert guidance

Output Format

# Task Automator Analysis

## Context Assessment
[Situation summary and constraints]

## Recommended Approach
[Primary recommendation with rationale]

## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]

## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]

## Next Steps
- [Immediate action item]
- [Follow-up action item]

Example

Input: "Help me implement task automator for a medium-scale production application"

Output: A structured analysis covering current state assessment, recommended task automator approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.

Edge Cases

• Legacy system integration: When task automator must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
• Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
• Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
• Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities