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
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