""" Base class for cron tasks. """ import logging import os import signal from abc import ABC, abstractmethod from datetime import date from typing import Optional, Dict, Any from django.db import transaction from django.conf import settings from django.utils import timezone from django_cronjob_utils.models import CronExecution from django_cronjob_utils.registry import ExecutionPattern from django_cronjob_utils.locks import DatabaseLock from django_cronjob_utils.exceptions import ValidationError, ConcurrentExecutionError from django_cronjob_utils.notifications import NotificationManager from django_cronjob_utils.hooks import CronjobHookContext, get_global_cronjob_hook_manager logger = logging.getLogger(__name__) class ExecutionResult: """Result of task execution.""" def __init__(self, success: bool = True, message: str = '', error_code: str = '', skipped: bool = False, reason: str = '', exception: Optional[Exception] = None, data: Optional[Dict[str, Any]] = None): self.success = success self.message = message self.error_code = error_code self.skipped = skipped self.reason = reason self.exception = exception self.data = data or {} class CronTask(ABC): """Abstract base class for all cron tasks.""" task_name: str = '' task_code: str = '' execution_pattern: str = ExecutionPattern.STANDARD retry_on_failure: bool = False max_retries: int = 0 retry_delay: int = 300 # seconds timeout: Optional[int] = None # seconds def __init__(self, execution_date: date, **options): """ Initialize cron task. Args: execution_date: Date for which the task should be executed **options: Additional options (e.g., force, rerun) """ self.execution_date = execution_date self.options = options self.execution: Optional[CronExecution] = None self._notification_manager = NotificationManager() self._hook_manager = get_global_cronjob_hook_manager() self._enable_hooks = options.get('enable_hooks', True) def validate(self): """Validate task input. Override in subclasses for custom validation.""" if not isinstance(self.execution_date, date): raise ValidationError(f"execution_date must be a date object, got {type(self.execution_date)}") def should_run(self) -> bool: """Check if task should run (duplicate prevention).""" force = self.options.get('force', False) if force: return True rerun = self.options.get('rerun', False) if rerun: return True if self.execution_pattern == ExecutionPattern.ALWAYS: return True if self.execution_pattern == ExecutionPattern.RERUN_ON_FAILURE: # Only skip if previous execution succeeded return not CronExecution.objects.filter( task_code=self.task_code, execution_date=self.execution_date, success=True ).exists() if self.execution_pattern == ExecutionPattern.RATE_LIMITED: # Check for concurrent execution (will be handled by lock) return True # STANDARD pattern: skip if already executed (only check completed executions) # Running executions are handled by the lock, not by should_run() return not CronExecution.objects.filter( task_code=self.task_code, execution_date=self.execution_date, completed=True ).exists() def acquire_lock(self): """Acquire database lock to prevent concurrent execution.""" return DatabaseLock(self.task_code, self.execution_date) def create_execution_record(self) -> CronExecution: """Create execution record in database.""" self.execution = CronExecution.objects.create( task_code=self.task_code, task_name=self.task_name, execution_date=self.execution_date, pid=os.getpid(), completed=False, success=False ) return self.execution def execute_with_timeout(self) -> ExecutionResult: """Execute with optional timeout.""" if self.timeout: return self._execute_with_signal_timeout() return self._execute() def _execute_with_signal_timeout(self) -> ExecutionResult: """Execute with signal-based timeout (Unix only).""" # Note: SIGALRM is only available on Unix systems # For Windows compatibility, consider using threading.Timer or similar if not hasattr(signal, 'SIGALRM'): logger.warning( f"Timeout not supported on this platform. " f"Task {self.task_name} will run without timeout." ) return self._execute() def timeout_handler(signum, frame): raise TimeoutError(f"Task execution exceeded timeout of {self.timeout} seconds") # Set up signal handler old_handler = signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(self.timeout) try: result = self._execute() finally: # Restore old handler and cancel alarm signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) return result def _execute(self) -> ExecutionResult: """Internal execution method.""" try: result = self.execute(self.execution_date) # Handle dict return value if isinstance(result, dict): return ExecutionResult( success=not result.get('error', False), message=result.get('message', ''), error_code=result.get('error_code', '') ) # Handle ExecutionResult return value if isinstance(result, ExecutionResult): return result # Default success return ExecutionResult( success=True, message=str(result) if result else 'Task completed successfully' ) except Exception as e: error_code = self.get_error_code(e) logger.exception(f"Task {self.task_name} execution failed: {e}") return ExecutionResult( success=False, message=str(e), error_code=error_code, exception=e ) def get_error_code(self, exception: Exception) -> str: """Get error code from exception. Override for custom error codes.""" return exception.__class__.__name__ def should_retry(self) -> bool: """Check if task should be retried.""" if not self.retry_on_failure: return False if not self.execution: return False if self.execution.retry_count >= self.max_retries: return False return True def schedule_retry(self): """Schedule retry execution. Override for custom retry logic.""" if self.execution: self.execution.increment_retry() logger.info( f"Scheduling retry {self.execution.retry_count + 1}/{self.max_retries} " f"for task {self.task_name} after {self.retry_delay} seconds" ) # Note: Actual retry scheduling would require Celery, Django-Q, or similar # For now, we just log and notify. The retry can be triggered manually # or via a separate management command that checks for failed executions. def notify_failure(self, result: ExecutionResult): """Notify stakeholders of failure.""" if self.execution: error_message = result.message or 'Unknown error' self._notification_manager.notify_failure(self.execution, error_message) def _create_hook_context(self, execution_id: Optional[int] = None, metadata: Optional[Dict[str, Any]] = None) -> CronjobHookContext: """Create hook context for current execution.""" return CronjobHookContext( operation='execute', task_code=self.task_code, task_name=self.task_name, execution_date=self.execution_date, execution_pattern=self.execution_pattern, execution_id=execution_id, # Can be None for pre-execution hooks retry_count=self.execution.retry_count if self.execution else 0, options=self.options, metadata=metadata or {} ) def _execute_pre_hooks(self, context: CronjobHookContext): """Execute PRE hooks before task execution.""" if not self._enable_hooks: return try: self._hook_manager.execute_pre_hooks(context) except Exception: # Let hook exceptions propagate raise def _execute_post_hooks(self, context: CronjobHookContext): """Execute POST hooks after task execution.""" if not self._enable_hooks: return [] return self._hook_manager.execute_post_hooks(context) def run(self) -> ExecutionResult: """Main execution method with full lifecycle management.""" # 1. Validate input self.validate() # 2. Check if should run (duplicate check) if not self.should_run(): return ExecutionResult( skipped=True, reason="Already executed" if self.execution_pattern != ExecutionPattern.ALWAYS else "Skipped" ) # 3. Execute PRE hooks (before lock acquisition to avoid DB records for rejected tasks) context = self._create_hook_context(execution_id=None) # No execution_id yet try: self._execute_pre_hooks(context) except Exception as e: # PRE hook rejected - no DB record created, just log and exit error_code = getattr(e, 'error_code', e.__class__.__name__) error_message = str(e) error_details = getattr(e, 'details', {}) logger.warning( f"Task {self.task_name} rejected by PRE hook: {error_code} - {error_message}" ) return ExecutionResult( success=False, message=error_message, error_code=error_code, data=error_details ) # 4. Acquire lock (prevent concurrent execution) execution_id = None try: with self.acquire_lock(): # 5. Create execution record self.execution = self.create_execution_record() execution_id = self.execution.id # Update context with execution_id (preserve metadata from PRE hooks) context = self._create_hook_context(execution_id, metadata=context.metadata) try: # 6. Execute with timeout result = self.execute_with_timeout() # 7. Update record self.execution.mark_completed( success=result.success, message=result.message, error_code=result.error_code ) # 8. Execute POST hooks post_hook_errors = self._execute_post_hooks(context) if post_hook_errors: logger.warning( f"Task {self.task_name} completed but POST hooks had errors: " f"{[e.error_code for e in post_hook_errors]}" ) # 9. Handle notifications if not result.success: self.notify_failure(result) # 10. Handle retry if not result.success and self.should_retry(): self.schedule_retry() # 11. Re-raise exception if execution failed if not result.success and result.exception: raise result.exception return result except (TimeoutError, Exception) as e: error_code = self.get_error_code(e) if not isinstance(e, TimeoutError) else "TIMEOUT" error_message = str(e) # Update execution record - this will be committed with the transaction if self.execution: self.execution.mark_failed( message=error_message, error_code=error_code ) self.notify_failure(ExecutionResult( success=False, message=error_message, error_code=error_code )) # Handle retry if self.should_retry(): self.schedule_retry() raise except ConcurrentExecutionError: # Re-raise ConcurrentExecutionError without modification raise except (TimeoutError, Exception) as e: # If execution record was created but transaction rolled back, save it in a new transaction if execution_id is not None: try: # Try to get the execution record - if it doesn't exist, transaction rolled back CronExecution.objects.get(pk=execution_id) except CronExecution.DoesNotExist: # Transaction rolled back, save execution record in a new transaction error_code = self.get_error_code(e) if not isinstance(e, TimeoutError) else "TIMEOUT" error_message = str(e) with transaction.atomic(): CronExecution.objects.create( task_code=self.task_code, task_name=self.task_name, execution_date=self.execution_date, pid=os.getpid(), completed=True, success=False, message=error_message, error_code=error_code, ended=timezone.now() ) # Re-raise the exception raise @abstractmethod def execute(self, date: date) -> dict: """ Business logic - must be implemented by subclasses. Args: date: Execution date Returns: dict with keys: - error: bool (True if error occurred) - message: str (status message) - error_code: str (optional error code) """ pass