FerroxLabs 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: ["team@example.com"])
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)
if args.dry_run:
config.dry_run = True
if args.verbose:
config.verbose = True
if args.output_dir:
config.output_dir = Path(args.output_dir)
return config
# ──────────────────────────────────────────────
# Logging Setup
# ──────────────────────────────────────────────
def setup_logging(verbose: bool = False) -> logging.Logger:
logger = logging.getLogger("daily_report")
logger.setLevel(logging.DEBUG if verbose else logging.INFO)
# Console handler
console = logging.StreamHandler(sys.stdout)
console.setFormatter(logging.Formatter(
'%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
))
logger.addHandler(console)
# File handler (rotated externally)
file_handler = logging.FileHandler('/var/log/daily_report.log')
file_handler.setFormatter(logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s: %(message)s'
))
logger.addHandler(file_handler)
return logger
# ──────────────────────────────────────────────
# Core Logic
# ──────────────────────────────────────────────
class ReportGenerator:
def __init__(self, config: Config, logger: logging.Logger):
self.config = config
self.logger = logger
self.stats = {"queries_run": 0, "rows_processed": 0, "errors": 0}
def run(self) -> bool:
"""Main execution flow. Returns True on success."""
start_time = time.time()
self.logger.info(f"Starting report generation for {self.config.report_date}")
if self.config.dry_run:
self.logger.info("[DRY RUN] No changes will be made")
try:
# Step 1: Validate prerequisites
self._validate()
# Step 2: Get data
data = self._fetch_data()
# Step 3: Generate report
report_path = self._generate_report(data)
# Step 4: Distribute
self._distribute(report_path)
elapsed = time.time() - start_time
self.logger.info(
f"Report completed in {elapsed:.1f}s. "
f"Stats: {self.stats}"
)
return True
except Exception as e:
self.stats["errors"] += 1
elapsed = time.time() - start_time
self.logger.error(f"Report generation FAILED after {elapsed:.1f}s: {e}", exc_info=True)
self._notify_failure(str(e))
return False
def _validate(self):
"""Check prerequisites before starting."""
self.config.output_dir.mkdir(parents=True, exist_ok=True)
# Verify database connectivity, SMTP reachability, etc.
self.logger.debug("Prerequisites validated")
def _fetch_data(self) -> dict:
"""Get data from database."""
self.logger.info("Fetching sales data...")
# ... database queries ...
self.stats["queries_run"] += 3
self.stats["rows_processed"] += 1500
return {"revenue": 50000, "orders": 230}
def _generate_report(self, data: dict) -> Path:
"""Generate the report file."""
report_path = self.config.output_dir / f"sales_{self.config.report_date}.pdf"
if self.config.dry_run:
self.logger.info(f"[DRY RUN] Would generate: {report_path}")
return report_path
# ... generate PDF ...
self.logger.info(f"Report generated: {report_path}")
return report_path
def _distribute(self, report_path: Path):
"""Send report to recipients."""
for recipient in self.config.recipients:
if self.config.dry_run:
self.logger.info(f"[DRY RUN] Would send to: {recipient}")
else:
# ... send email ...
self.logger.info(f"Report sent to: {recipient}")
def _notify_failure(self, error_message: str):
"""Send failure notification."""
self.logger.info(f"Sending failure notification: {error_message}")
# ... send alert via Slack/PagerDuty/email ...
# ──────────────────────────────────────────────
# Entry Point
# ──────────────────────────────────────────────
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Generate daily sales report")
parser.add_argument("--date", help="Report date (YYYY-MM-DD)")
parser.add_argument("--dry-run", action="store_true", help="Preview without executing")
parser.add_argument("--verbose", "-v", action="store_true", help="Debug logging")
parser.add_argument("--output-dir", help="Output directory path")
parser.add_argument("--config", help="Path to config YAML file")
return parser.parse_args()
def main():
args = parse_args()
config = Config.from_args(args)
logger = setup_logging(verbose=config.verbose)
generator = ReportGenerator(config, logger)
success = generator.run()
sys.exit(0 if success else 1)
if __name__ == "__main__":
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))
def fetch_data_from_api(url):
response = httpx.get(url, timeout=30)
response.raise_for_status()
return 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 not processor.should_process(record.id):
logger.debug(f"Skipping already-processed: {record.id}")
continue
migrate_record(record)
processor.mark_processed(record.id)
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'] = 'automation@example.com'
msg['To'] = ', '.join(recipients)
with smtplib.SMTP('smtp.example.com', 587) as server:
server.starttls()
server.login('automation@example.com', '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: ["team@example.com", "alerts@example.com"]
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
- Always implement dry-run mode: Test changes safely before executing
- Make scripts idempotent: Safe to run multiple times with same result
- Use structured logging: Include timestamps, levels, and context
- Fail loudly: Send notifications on failure, exit with non-zero code
- Use configuration files: Avoid hardcoded values
- Handle signals gracefully: Respond to SIGTERM for clean shutdown
- Add timeouts: Every external call should have a timeout
- Document everything: Purpose, dependencies, scheduling, recovery procedures
- Version control your scripts: Track changes like any other code
- 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