workflow-automation

What this skill does

# Workflow Automation

## Business Case

### Problem Statement
Data workflow challenges:
- Manual repetitive tasks
- Data inconsistency between systems
- Error-prone manual processes
- Lack of audit trails

### Solution
Automated workflow system for construction data pipelines with task dependencies, scheduling, and monitoring.

## Technical Implementation

```python
import pandas as pd
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json


class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


class TriggerType(Enum):
    MANUAL = "manual"
    SCHEDULED = "scheduled"
    EVENT = "event"
    DEPENDENCY = "dependency"


class ScheduleInterval(Enum):
    HOURLY = "hourly"
    DAILY = "daily"
    WEEKLY = "weekly"
    MONTHLY = "monthly"


@dataclass
class WorkflowTask:
    task_id: str
    name: str
    task_type: str  # extract, transform, load, notify, validate
    config: Dict[str, Any] = field(default_factory=dict)
    dependencies: List[str] = field(default_factory=list)
    retries: int = 3
    timeout_minutes: int = 30


@dataclass
class TaskExecution:
    task_id: str
    status: TaskStatus
    start_time: datetime
    end_time: Optional[datetime] = None
    output: Any = None
    error: str = ""
    attempt: int = 1


@dataclass
class Workflow:
    workflow_id: str
    name: str
    description: str
    tasks: List[WorkflowTask] = field(default_factory=list)
    trigger_type: TriggerType = TriggerType.MANUAL
    schedule: Optional[ScheduleInterval] = None


class WorkflowAutomation:
    """Automate construction data workflows and ETL pipelines."""

    def __init__(self, project_name: str):
        self.project_name = project_name
        self.workflows: Dict[str, Workflow] = {}
        self.executions: Dict[str, List[TaskExecution]] = {}
        self.task_handlers: Dict[str, Callable] = {}
        self._register_default_handlers()

    def _register_default_handlers(self):
        """Register default task handlers."""

        self.task_handlers['extract_csv'] = self._extract_csv
        self.task_handlers['extract_excel'] = self._extract_excel
        self.task_handlers['transform_filter'] = self._transform_filter
        self.task_handlers['transform_aggregate'] = self._transform_aggregate
        self.task_handlers['transform_join'] = self._transform_join
        self.task_handlers['load_csv'] = self._load_csv
        self.task_handlers['validate_schema'] = self._validate_schema
        self.task_handlers['notify_email'] = self._notify_email

    def register_handler(self, task_type: str, handler: Callable):
        """Register custom task handler."""
        self.task_handlers[task_type] = handler

    def create_workflow(self, workflow_id: str, name: str,
                        description: str = "") -> Workflow:
        """Create new workflow."""

        workflow = Workflow(
            workflow_id=workflow_id,
            name=name,
            description=description
        )
        self.workflows[workflow_id] = workflow
        self.executions[workflow_id] = []
        return workflow

    def add_task(self, workflow_id: str, task: WorkflowTask):
        """Add task to workflow."""

        if workflow_id in self.workflows:
            self.workflows[workflow_id].tasks.append(task)

    def set_schedule(self, workflow_id: str, interval: ScheduleInterval):
        """Set workflow schedule."""

        if workflow_id in self.workflows:
            self.workflows[workflow_id].trigger_type = TriggerType.SCHEDULED
            self.workflows[workflow_id].schedule = interval

    def get_execution_order(self, workflow_id: str) -> List[str]:
        """Get task execution order based on dependencies (topological sort)."""

        if workflow_id not in self.workflows:
            return []

        workflow = self.workflows[workflow_id]
        tasks = {t.task_id: t for t in workflow.tasks}

        # Build dependency graph
        in_degree = {t.task_id: 0 for t in workflow.tasks}
        graph = {t.task_id: [] for t in workflow.tasks}

        for task in workflow.tasks:
            for dep in task.dependencies:
                if dep in graph:
                    graph[dep].append(task.task_id)
                    in_degree[task.task_id] += 1

        # Topological sort
        queue = [t for t in in_degree if in_degree[t] == 0]
        order = []

        while queue:
            task_id = queue.pop(0)
            order.append(task_id)

            for dependent in graph[task_id]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)

        return order

    def execute_workflow(self, workflow_id: str,
                         context: Dict[str, Any] = None) -> Dict[str, Any]:
        """Execute workflow."""

        if workflow_id not in self.workflows:
            return {'error': f'Workflow {workflow_id} not found'}

        workflow = self.workflows[workflow_id]
        context = context or {}
        execution_results = []
        task_outputs = {}

        execution_order = self.get_execution_order(workflow_id)
        start_time = datetime.now()

        for task_id in execution_order:
            task = next(t for t in workflow.tasks if t.task_id == task_id)

            # Check dependencies
            deps_success = all(
                task_outputs.get(dep, {}).get('status') == TaskStatus.SUCCESS
                for dep in task.dependencies
            )

            if not deps_success:
                execution = TaskExecution(
                    task_id=task_id,
                    status=TaskStatus.SKIPPED,
                    start_time=datetime.now(),
                    end_time=datetime.now(),
                    error="Dependencies not met"
                )
            else:
                execution = self._execute_task(task, context, task_outputs)

            task_outputs[task_id] = {
                'status': execution.status,
                'output': execution.output
            }
            execution_results.append(execution)
            self.executions[workflow_id].append(execution)

        end_time = datetime.now()
        success_count = sum(1 for e in execution_results if e.status == TaskStatus.SUCCESS)

        return {
            'workflow_id': workflow_id,
            'workflow_name': workflow.name,
            'start_time': start_time.isoformat(),
            'end_time': end_time.isoformat(),
            'duration_seconds': (end_time - start_time).total_seconds(),
            'total_tasks': len(execution_results),
            'successful_tasks': success_count,
            'failed_tasks': sum(1 for e in execution_results if e.status == TaskStatus.FAILED),
            'skipped_tasks': sum(1 for e in execution_results if e.status == TaskStatus.SKIPPED),
            'overall_status': 'success' if success_count == len(execution_results) else 'failed',
            'task_results': [
                {
                    'task_id': e.task_id,
                    'status': e.status.value,
                    'error': e.error
                }
                for e in execution_results
            ]
        }

    def _execute_task(self, task: WorkflowTask, context: Dict[str, Any],
                      task_outputs: Dict[str, Any]) -> TaskExecution:
        """Execute single task."""

        execution = TaskExecution(
            task_id=task.task_id,
            status=TaskStatus.RUNNING,
            start_time=datetime.now()
        )

        handler = self.task_handlers.get(task.task_type)

        if not handler:
            execution.status = TaskStatus.FAILED
            execution.error = f"No handler for task type: {task.task_type}"
            execution.end_time = datetime.now()
            return execution