""" Dependency Manager for Python Scheduler Handles job chaining and execution order with configurable error handling """ import time from typing import List, Dict, Any, Optional from datetime import datetime from loguru import logger from core.database import SchedulerDatabase from daemon.job_executor import JobExecutor class DependencyManager: """Manages job dependencies and execution chains""" def __init__(self, db: SchedulerDatabase, executor: JobExecutor): self.db = db self.executor = executor logger.info("DependencyManager initialized") def execute_job_group(self, group_id: int, triggered_by: str = 'scheduler') -> Dict[str, Any]: """ Execute a job group with dependency chain Args: group_id: Job group ID to execute triggered_by: Who/what triggered execution Returns: Dictionary with execution results """ group = self.db.get_job_group(group_id) if not group: logger.error(f"Job group {group_id} not found") return { 'success': False, 'error': f"Job group {group_id} not found" } members = self.db.get_job_group_members(group_id) if not members: logger.warning(f"Job group '{group['name']}' has no members") return { 'success': False, 'error': 'No jobs in group' } logger.info(f"Starting execution of job group '{group['name']}' ({len(members)} jobs)") error_handling = group['error_handling'] parent_execution_id = None execution_results = [] group_start_time = datetime.now() for i, member in enumerate(members, 1): job_id = member['job_id'] job_name = member['job_name'] logger.info(f"Executing job {i}/{len(members)}: '{job_name}' (ID {job_id})") # Execute job execution_id = self.executor.execute_job( job_id=job_id, triggered_by=f"group:{group['name']}", group_id=group_id, parent_execution_id=parent_execution_id ) if not execution_id: logger.error(f"Failed to queue job '{job_name}'") execution_results.append({ 'job_id': job_id, 'job_name': job_name, 'execution_id': None, 'status': 'failed', 'error': 'Failed to queue' }) # Handle error if error_handling == 'stop_on_error': logger.info("Stopping group execution due to queue failure") break elif error_handling == 'skip_remaining': logger.info("Skipping remaining jobs in group") break # else: continue to next job continue # Wait for completion final_status = self._wait_for_completion(execution_id, job_name) # Get execution details execution = self.db.get_execution(execution_id) execution_results.append({ 'job_id': job_id, 'job_name': job_name, 'execution_id': execution_id, 'status': final_status, 'duration': execution.get('duration_seconds'), 'exit_code': execution.get('exit_code') }) # Handle failure if final_status not in ['completed']: logger.error(f"Job '{job_name}' ended with status: {final_status}") if error_handling == 'stop_on_error': logger.info("Stopping group execution due to error") break elif error_handling == 'skip_remaining': logger.info("Skipping remaining jobs in group") break # else: continue to next job parent_execution_id = execution_id # Calculate group execution time group_duration = (datetime.now() - group_start_time).total_seconds() # Determine overall success all_completed = all(r['status'] == 'completed' for r in execution_results) any_failed = any(r['status'] in ['failed', 'timeout'] for r in execution_results) logger.info(f"Job group '{group['name']}' finished in {group_duration:.1f}s - " f"Completed: {sum(1 for r in execution_results if r['status'] == 'completed')}/{len(execution_results)}") return { 'success': all_completed, 'group_id': group_id, 'group_name': group['name'], 'total_jobs': len(members), 'executed_jobs': len(execution_results), 'completed_jobs': sum(1 for r in execution_results if r['status'] == 'completed'), 'failed_jobs': sum(1 for r in execution_results if r['status'] in ['failed', 'timeout']), 'duration_seconds': group_duration, 'executions': execution_results } def _wait_for_completion(self, execution_id: int, job_name: str, poll_interval: float = 1.0) -> str: """ Wait for job execution to complete Args: execution_id: Execution ID to monitor job_name: Job name for logging poll_interval: How often to check status (seconds) Returns: Final status ('completed', 'failed', 'timeout', 'cancelled') """ logger.debug(f"Waiting for execution {execution_id} ('{job_name}') to complete") while True: status = self.db.get_execution_status(execution_id) if status in ['completed', 'failed', 'timeout', 'cancelled']: logger.debug(f"Execution {execution_id} finished with status: {status}") return status time.sleep(poll_interval) def create_job_chain(self, name: str, job_ids: List[int], description: str = None, error_handling: str = 'stop_on_error') -> int: """ Create a job group (chain) from a list of job IDs Args: name: Group name job_ids: List of job IDs in execution order description: Group description error_handling: 'stop_on_error', 'continue', or 'skip_remaining' Returns: Group ID """ # Validate jobs exist for job_id in job_ids: job = self.db.get_job(job_id) if not job: raise ValueError(f"Job ID {job_id} not found") # Create group group_id = self.db.create_job_group(name, description, error_handling) # Add members for order, job_id in enumerate(job_ids, 1): self.db.add_job_to_group(group_id, job_id, order) logger.info(f"Created job chain '{name}' with {len(job_ids)} jobs") return group_id def get_group_execution_summary(self, group_id: int, limit: int = 10) -> List[Dict[str, Any]]: """ Get execution summary for a job group Args: group_id: Job group ID limit: Number of recent executions to retrieve Returns: List of execution summaries """ # Get recent executions for this group with self.db._get_connection() as conn: cursor = conn.execute(""" SELECT e.id, e.job_id, j.name as job_name, e.status, e.start_time, e.end_time, e.duration_seconds, e.exit_code, e.triggered_by FROM job_executions e JOIN jobs j ON e.job_id = j.id WHERE e.group_id = ? ORDER BY e.start_time DESC LIMIT ? """, (group_id, limit)) executions = [dict(row) for row in cursor.fetchall()] # Group by start_time (same batch) batches = {} for ex in executions: start_key = ex['start_time'][:16] # Group by minute if start_key not in batches: batches[start_key] = [] batches[start_key].append(ex) summaries = [] for start_key, batch_execs in sorted(batches.items(), reverse=True): batch_execs_sorted = sorted(batch_execs, key=lambda x: x['start_time']) first_exec = batch_execs_sorted[0] last_exec = batch_execs_sorted[-1] total_duration = 0 for ex in batch_execs_sorted: if ex['duration_seconds']: total_duration += ex['duration_seconds'] all_completed = all(ex['status'] == 'completed' for ex in batch_execs_sorted) summaries.append({ 'start_time': first_exec['start_time'], 'end_time': last_exec['end_time'], 'total_duration': total_duration, 'jobs_executed': len(batch_execs_sorted), 'all_completed': all_completed, 'executions': batch_execs_sorted }) return summaries def validate_job_group(self, group_id: int) -> Dict[str, Any]: """ Validate a job group configuration Args: group_id: Job group ID Returns: Validation results """ group = self.db.get_job_group(group_id) if not group: return { 'valid': False, 'errors': ['Group not found'] } members = self.db.get_job_group_members(group_id) errors = [] warnings = [] if not members: errors.append('Group has no members') for member in members: job = self.db.get_job(member['job_id']) if not job: errors.append(f"Job ID {member['job_id']} not found") elif not job['enabled']: warnings.append(f"Job '{job['name']}' is disabled") return { 'valid': len(errors) == 0, 'errors': errors, 'warnings': warnings, 'total_jobs': len(members) }