# Django Cronjob Utils - Usage Guide

This document provides practical implementation examples and code snippets for using the django-cronjob-utils package.

## Quick Start

### 1. Installation

```bash
pip install django-cronjob-utils
```

### 2. Add to Django Settings

```python
# settings.py
INSTALLED_APPS = [
    # ...
    'django_cronjob_utils',
]

# Optional: Configure notifications
CRONJOB_UTILS = {
    'NOTIFICATIONS': {
        'on_failure': ['email'],
        'email': {
            'recipients': ['admin@example.com'],
        },
    },
}
```

### 3. Run Migrations

```bash
python manage.py migrate django_cronjob_utils
```

### 4. Create Your First Task

```python
# myapp/tasks.py
from django_cronjob_utils import CronTask, register_task
from datetime import date

@register_task('calc-commission', 'A001')
class CalcCommissionTask(CronTask):
    def execute(self, date: date) -> dict:
        # Your business logic here
        return {'error': False, 'message': 'Success'}
```

### 5. Set Up Crontab

```bash
# Edit crontab
crontab -e

# Add entry (runs daily at 1 AM)
0 1 * * * cd /path/to/project && /path/to/venv/bin/python manage.py run_cron_task calc-commission $(date +\%Y-\%m-\%d)
```

## Task Implementation Patterns

### Pattern 1: Simple Task (Standard Execution)

```python
@register_task('daily-report', 'A001')
class DailyReportTask(CronTask):
    """Generate daily report - runs once per day"""
    
    def execute(self, date: date) -> dict:
        try:
            report_service = ReportService()
            report = report_service.generate_daily_report(date)
            
            return {
                'error': False,
                'message': f'Report generated: {report.filename}'
            }
        except Exception as e:
            return {
                'error': True,
                'message': str(e),
                'error_code': 'REPORT_GENERATION_ERROR'
            }
```

### Pattern 2: Task with Retry on Failure

```python
@register_task('sync-external-data', 'A002',
               execution_pattern=ExecutionPattern.RERUN_ON_FAILURE,
               retry_on_failure=True,
               max_retries=3,
               retry_delay=300)  # 5 minutes between retries
class SyncExternalDataTask(CronTask):
    """Sync data from external API - retries on failure"""
    
    def execute(self, date: date) -> dict:
        api_client = ExternalAPIClient()
        
        try:
            result = api_client.sync(date)
            
            if result.success:
                return {
                    'error': False,
                    'message': f'Synced {result.record_count} records'
                }
            else:
                return {
                    'error': True,
                    'message': result.error_message,
                    'error_code': 'API_SYNC_ERROR'
                }
        except APIConnectionError as e:
            return {
                'error': True,
                'message': f'API connection failed: {str(e)}',
                'error_code': 'API_CONNECTION_ERROR'
            }
```

### Pattern 3: Always Execute (No Duplicate Check)

```python
@register_task('update-cache', 'A003',
               execution_pattern=ExecutionPattern.ALWAYS)
class UpdateCacheTask(CronTask):
    """Update cache - safe to run multiple times"""
    
    def execute(self, date: date) -> dict:
        cache_service = CacheService()
        cache_service.refresh_all()
        
        return {
            'error': False,
            'message': 'Cache updated successfully'
        }
```

### Pattern 4: Rate-Limited Task (Prevent Concurrent Execution)

```python
@register_task('process-large-batch', 'A004',
               execution_pattern=ExecutionPattern.RATE_LIMITED,
               timeout=7200)  # 2 hour timeout
class ProcessLargeBatchTask(CronTask):
    """Process large batch - prevents concurrent execution"""
    
    def execute(self, date: date) -> dict:
        batch_processor = BatchProcessor()
        processed_count = 0
        error_count = 0
        
        while True:
            batch = batch_processor.get_next_batch(date)
            if not batch:
                break
            
            try:
                batch_processor.process(batch)
                processed_count += batch.size
            except Exception as e:
                error_count += 1
                logger.error(f"Error processing batch: {e}")
        
        if error_count > 0:
            return {
                'error': True,
                'message': f'Processed {processed_count} records with {error_count} errors',
                'error_code': 'BATCH_PROCESSING_ERROR'
            }
        
        return {
            'error': False,
            'message': f'Successfully processed {processed_count} records'
        }
```

### Pattern 5: Task with Custom Validation

```python
@register_task('process-payment', 'A005')
class ProcessPaymentTask(CronTask):
    """Process payments - requires custom validation"""
    
    def validate(self):
        """Custom validation logic"""
        super().validate()  # Call parent validation
        
        # Check if it's a business day
        if self.execution_date.weekday() >= 5:  # Saturday or Sunday
            raise ValidationError("Payments can only be processed on weekdays")
        
        # Check if it's not a holiday
        if HolidayService.is_holiday(self.execution_date):
            raise ValidationError(f"{self.execution_date} is a holiday")
    
    def execute(self, date: date) -> dict:
        payment_service = PaymentService()
        result = payment_service.process_daily_payments(date)
        
        return {
            'error': False,
            'message': f'Processed {result.count} payments'
        }
```

### Pattern 6: Task with Custom Error Handling

```python
@register_task('import-data', 'A006')
class ImportDataTask(CronTask):
    """Import data from file - custom error handling"""
    
    def execute(self, date: date) -> dict:
        importer = DataImporter()
        
        try:
            result = importer.import_from_file(date)
            
            if result.has_warnings:
                # Log warnings but don't fail
                logger.warning(f"Import completed with warnings: {result.warnings}")
            
            return {
                'error': False,
                'message': f'Imported {result.record_count} records'
            }
        except FileNotFoundError:
            return {
                'error': True,
                'message': f'Import file not found for {date}',
                'error_code': 'FILE_NOT_FOUND'
            }
        except ValidationError as e:
            return {
                'error': True,
                'message': f'Data validation failed: {str(e)}',
                'error_code': 'VALIDATION_ERROR'
            }
        except Exception as e:
            # Unexpected error
            logger.exception(f"Unexpected error during import: {e}")
            return {
                'error': True,
                'message': f'Unexpected error: {str(e)}',
                'error_code': 'UNEXPECTED_ERROR'
            }
```

## Advanced Usage

### Custom Execution Pattern

```python
from django_cronjob_utils import CronTask, ExecutionPattern

class CustomExecutionPattern(ExecutionPattern):
    """Custom pattern: Only run on weekdays"""
    
    @staticmethod
    def should_run(task_code: str, execution_date: date) -> bool:
        # Check if already executed
        if CronExecution.objects.filter(
            task_code=task_code,
            execution_date=execution_date
        ).exists():
            return False
        
        # Only run on weekdays
        if execution_date.weekday() >= 5:
            return False
        
        return True

@register_task('weekday-task', 'A007',
               execution_pattern=CustomExecutionPattern)
class WeekdayTask(CronTask):
    def execute(self, date: date) -> dict:
        return {'error': False, 'message': 'Task executed'}
```

### Task with Dependencies

```python
@register_task('dependent-task', 'A008')
class DependentTask(CronTask):
    """Task that depends on another task completing successfully"""
    
    def should_run(self) -> bool:
        # Check if dependency completed successfully
        dependency_completed = CronExecution.objects.filter(
            task_code='A001',  # Dependency task code
            execution_date=self.execution_date,
            success=True
        ).exists()
        
        if not dependency_completed:
            logger.info(f"Dependency A001 not completed for {self.execution_date}")
            return False
        
        # Standard duplicate check
        return super().should_run()
    
    def execute(self, date: date) -> dict:
        # Your logic here
        return {'error': False, 'message': 'Task executed'}
```

### Task with Progress Tracking

```python
@register_task('long-running-task', 'A009',
               timeout=10800)  # 3 hours
class LongRunningTask(CronTask):
    """Long-running task with progress tracking"""
    
    def execute(self, date: date) -> dict:
        processor = LongRunningProcessor()
        total_items = processor.get_total_count(date)
        processed = 0
        
        for item in processor.get_items(date):
            try:
                processor.process_item(item)
                processed += 1
                
                # Update progress every 100 items
                if processed % 100 == 0:
                    self.execution.message = f'Progress: {processed}/{total_items}'
                    self.execution.save(update_fields=['message'])
                    
            except Exception as e:
                logger.error(f"Error processing item {item.id}: {e}")
        
        return {
            'error': False,
            'message': f'Processed {processed}/{total_items} items'
        }
```

## Notification Configuration

### Email Notifications

```python
# settings.py
CRONJOB_UTILS = {
    'NOTIFICATIONS': {
        'on_failure': ['email'],
        'email': {
            'recipients': ['admin@example.com', 'devops@example.com'],
            'from_email': 'noreply@example.com',
            'subject_template': 'Cronjob Failed: {task_name}',
        },
    },
}
```

### Slack Notifications

```python
# settings.py
CRONJOB_UTILS = {
    'NOTIFICATIONS': {
        'on_failure': ['slack'],
        'slack': {
            'webhook_url': 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL',
            'channel': '#cronjobs',
            'username': 'Cronbot',
        },
    },
}
```

### Telegram Notifications

```python
# settings.py
CRONJOB_UTILS = {
    'NOTIFICATIONS': {
        'on_failure': ['telegram'],
        'telegram': {
            'bot_token': 'your-bot-token',  # Get from @BotFather
            'chat_id': 'your-chat-id',      # Your Telegram user ID or group chat ID
        },
    },
}
```

**Setting up Telegram Bot**:

1. **Create a Bot**:
   - Open Telegram and search for `@BotFather`
   - Send `/newbot` command
   - Follow instructions to create your bot
   - Save the bot token provided

2. **Get Chat ID**:
   - For personal notifications: Search for `@userinfobot` and send `/start` - it will show your user ID
   - For group notifications: Add `@userinfobot` to your group and it will show the group chat ID
   - Alternatively, send a message to your bot and visit: `https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getUpdates` to find the chat_id

3. **Example Configuration**:
```python
CRONJOB_UTILS = {
    'NOTIFICATIONS': {
        'on_failure': ['telegram'],
        'telegram': {
            'bot_token': '123456789:ABCdefGHIjklMNOpqrsTUVwxyz',
            'chat_id': '123456789',  # Your user ID or group chat ID
        },
    },
}
```

**Multiple Notification Backends**:

You can configure multiple notification backends to receive alerts via different channels:

```python
CRONJOB_UTILS = {
    'NOTIFICATIONS': {
        'on_failure': ['email', 'slack', 'telegram'],
        'email': {
            'recipients': ['admin@example.com'],
        },
        'slack': {
            'webhook_url': 'https://hooks.slack.com/services/...',
            'channel': '#cronjobs',
        },
        'telegram': {
            'bot_token': 'your-bot-token',
            'chat_id': 'your-chat-id',
        },
    },
}
```

### Custom Notification Backend

```python
from django_cronjob_utils.notifications import NotificationBackend

class CustomNotificationBackend(NotificationBackend):
    """Custom notification backend (e.g., PagerDuty, SMS)"""
    
    def notify_failure(self, execution, error_message):
        # Your notification logic
        pagerduty_client = PagerDutyClient()
        pagerduty_client.trigger_incident(
            summary=f"Cronjob {execution.task_name} failed",
            details=error_message
        )

# Register in settings
CRONJOB_UTILS = {
    'NOTIFICATIONS': {
        'on_failure': ['custom'],
        'custom': {
            'backend': 'myapp.notifications.CustomNotificationBackend',
            'api_key': 'your-api-key',
        },
    },
}
```

## Monitoring and Querying

### Find Recent Failures

```python
from django_cronjob_utils.models import CronExecution
from datetime import date, timedelta

# Failures in last 7 days
failures = CronExecution.objects.filter(
    success=False,
    execution_date__gte=date.today() - timedelta(days=7)
).order_by('-started')

for failure in failures:
    print(f"{failure.task_name} failed on {failure.execution_date}: {failure.message}")
```

### Find Stuck Jobs

```python
from django.utils import timezone
from datetime import timedelta

# Jobs running longer than 1 hour
stuck_jobs = CronExecution.objects.filter(
    completed=False,
    started__lt=timezone.now() - timedelta(hours=1)
)

for job in stuck_jobs:
    duration = timezone.now() - job.started
    print(f"{job.task_name} stuck for {duration}")
```

### Task Success Rate

```python
from django.db.models import Count, Q, Avg
from datetime import date, timedelta

stats = CronExecution.objects.filter(
    execution_date__gte=date.today() - timedelta(days=30)
).values('task_name').annotate(
    total=Count('id'),
    successful=Count('id', filter=Q(success=True)),
    failed=Count('id', filter=Q(success=False)),
    avg_duration=Avg('ended' - 'started')
).order_by('task_name')

for stat in stats:
    success_rate = (stat['successful'] / stat['total']) * 100 if stat['total'] > 0 else 0
    print(f"{stat['task_name']}: {success_rate:.1f}% success rate")
```

### Export Execution History

```python
import csv
from django_cronjob_utils.models import CronExecution

def export_execution_history(start_date, end_date, output_file):
    executions = CronExecution.objects.filter(
        execution_date__gte=start_date,
        execution_date__lte=end_date
    ).order_by('-started')
    
    with open(output_file, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([
            'Task Name', 'Task Code', 'Execution Date', 'Started', 'Ended',
            'Success', 'Message', 'Error Code', 'Duration (seconds)'
        ])
        
        for exec in executions:
            duration = exec.duration if exec.duration else ''
            writer.writerow([
                exec.task_name,
                exec.task_code,
                exec.execution_date,
                exec.started,
                exec.ended,
                exec.success,
                exec.message,
                exec.error_code,
                duration
            ])
```

## Testing

### Unit Test Example

```python
from django.test import TestCase
from django_cronjob_utils import CronTask, register_task
from django_cronjob_utils.models import CronExecution
from datetime import date

@register_task('test-task', 'T001')
class TestTask(CronTask):
    def execute(self, date: date) -> dict:
        return {'error': False, 'message': 'Test success'}

class CronTaskTestCase(TestCase):
    def setUp(self):
        self.test_date = date(2024, 1, 15)
    
    def test_task_execution(self):
        """Test successful task execution"""
        task = TestTask(self.test_date)
        result = task.run()
        
        self.assertTrue(result.success)
        self.assertEqual(result.message, 'Test success')
        
        # Check database record
        execution = CronExecution.objects.get(
            task_code='T001',
            execution_date=self.test_date
        )
        self.assertTrue(execution.success)
        self.assertTrue(execution.completed)
    
    def test_duplicate_prevention(self):
        """Test that duplicate execution is prevented"""
        task1 = TestTask(self.test_date)
        result1 = task1.run()
        self.assertTrue(result1.success)
        
        # Second execution should be skipped
        task2 = TestTask(self.test_date)
        result2 = task2.run()
        self.assertTrue(result2.skipped)
        self.assertEqual(result2.reason, "Already executed")
    
    def test_retry_on_failure(self):
        """Test retry mechanism"""
        @register_task('failing-task', 'T002',
                      retry_on_failure=True,
                      max_retries=2)
        class FailingTask(CronTask):
            call_count = 0
            
            def execute(self, date: date) -> dict:
                FailingTask.call_count += 1
                if FailingTask.call_count < 3:
                    return {'error': True, 'message': 'Simulated failure'}
                return {'error': False, 'message': 'Success after retries'}
        
        task = FailingTask(self.test_date)
        # Note: Actual retry scheduling would need Celery or similar
        # This test just verifies the retry logic
        self.assertTrue(task.should_retry())
```

### Integration Test Example

```python
from django.test import TestCase
from django.core.management import call_command
from io import StringIO
from datetime import date

class ManagementCommandTest(TestCase):
    def test_run_cron_task_command(self):
        """Test management command execution"""
        out = StringIO()
        execution_date = date.today().isoformat()
        
        # Register a test task first
        # (In real scenario, this would be in your app's tasks.py)
        
        call_command(
            'run_cron_task',
            'test-task',
            execution_date,
            stdout=out
        )
        
        self.assertIn('completed successfully', out.getvalue())
```

## Troubleshooting

### Common Issues

#### 1. Task Not Found

**Error**: `TaskNotFoundError: Task 'my-task' not found`

**Solution**: Ensure task is registered before running command. Import the module containing the task:

```python
# In your app's __init__.py or ready() method
from myapp import tasks  # This registers the tasks
```

#### 2. Concurrent Execution Error

**Error**: `ConcurrentExecutionError: Task already running`

**Solution**: This is expected behavior. The task is already running. Check for stuck jobs:

```python
from django_cronjob_utils.models import CronExecution
from django.utils import timezone
from datetime import timedelta

stuck = CronExecution.objects.filter(
    completed=False,
    started__lt=timezone.now() - timedelta(hours=1)
)
```

#### 3. Timeout Issues

**Error**: Task times out even though it should complete

**Solution**: Increase timeout or optimize task:

```python
@register_task('slow-task', 'A010',
               timeout=14400)  # 4 hours instead of default 1 hour
class SlowTask(CronTask):
    pass
```

#### 4. Notification Not Working

**Error**: No notifications received on failure

**Solution**: Check settings and ensure notification backends are properly configured:

```python
# Test notification manually
from django_cronjob_utils.notifications import NotificationManager

manager = NotificationManager()
manager.notify_failure(execution, "Test error message")
```

#### 5. Telegram Notification Issues

**Error**: Telegram notifications not being sent

**Solution**: 
- Verify bot token is correct (from @BotFather)
- Verify chat_id is correct (use @userinfobot or check getUpdates API)
- Ensure bot has permission to send messages to the chat
- Check that requests library is installed: `pip install requests`

**Test Telegram Bot**:
```python
import requests

bot_token = 'your-bot-token'
chat_id = 'your-chat-id'
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"

payload = {
    'chat_id': chat_id,
    'text': 'Test message'
}

response = requests.post(url, json=payload)
print(response.json())
```

## Best Practices

1. **Always return proper error codes**: Helps with monitoring and debugging
2. **Use descriptive task names**: Makes monitoring easier
3. **Set appropriate timeouts**: Prevents stuck jobs
4. **Enable retry for transient failures**: Improves reliability
5. **Monitor execution history**: Regular checks for failures and stuck jobs
6. **Test tasks before deploying**: Use unit tests and staging environment
7. **Document task purpose**: Add docstrings explaining what each task does
8. **Use appropriate execution patterns**: Choose the right pattern for each task
9. **Clean up old records**: Archive or delete old execution records periodically
10. **Set up alerts**: Configure notifications for critical tasks
11. **Use multiple notification channels**: Email + Telegram or Slack for redundancy
12. **Test notification backends**: Verify notifications work before deploying to production

## Additional Resources

- Package documentation: See architecture.md
- Django management commands: https://docs.djangoproject.com/en/stable/howto/custom-management-commands/
- Django signals: For custom event handling
- Celery: For advanced task scheduling and retry mechanisms
- Telegram Bot API: https://core.telegram.org/bots/api