task-patterns

What this skill does

# Task Patterns Skill

Production-ready Celery task templates with comprehensive error handling, retry mechanisms, rate limiting, and custom behavior patterns.

## Overview

This skill provides battle-tested templates and patterns for building robust Celery tasks. Each template demonstrates production-ready code with proper error handling, logging, retry logic, and security best practices.

## Core Patterns

### 1. Basic Task Template

**Template**: `templates/basic-task.py`

Simple task with standard error handling and logging.

**Use when**:
- Creating your first Celery task
- Need straightforward task execution
- No complex retry or limiting requirements

**Features**:
- Proper logging setup
- Basic error handling
- Synchronous and asynchronous execution examples
- Clear docstrings

### 2. Retry Task Template

**Template**: `templates/retry-task.py`

Tasks with automatic retry mechanisms and exponential backoff.

**Use when**:
- Calling external APIs that may fail transiently
- Database operations that could timeout
- Network operations prone to connection issues

**Features**:
- Automatic retry with `autoretry_for`
- Exponential backoff with jitter
- Manual retry control
- Fixed delay retries
- Configurable max retries

**Example**: See `examples/task-with-retries.md` for complete implementation guide

### 3. Rate Limited Task Template

**Template**: `templates/rate-limited-task.py`

Tasks with rate limiting to control execution speed.

**Use when**:
- Respecting external API rate limits
- Protecting database from overload
- Controlling resource consumption
- Meeting SLA requirements

**Features**:
- Per-second, per-minute, per-hour limits
- Batch processing with rate control
- Large dataset handling
- Dynamic rate limit adjustment

**Example**: See `examples/rate-limiting.md` for complete guide

### 4. Time Limited Task Template

**Template**: `templates/time-limited-task.py`

Tasks with soft and hard time limits to prevent runaway execution.

**Use when**:
- Long-running operations that could hang
- Operations with external dependencies
- Tasks that must complete within timeframe
- Preventing resource exhaustion

**Features**:
- Soft time limit (catchable)
- Hard time limit (force kill)
- Graceful timeout handling
- Progress saving on timeout
- Combined with retry logic

### 5. Custom Task Class Template

**Template**: `templates/custom-task-class.py`

Custom task base classes with specialized behavior.

**Use when**:
- Need database connection pooling
- Want to cache task results
- Require metrics tracking
- Need shared resources across tasks
- Implementing lifecycle hooks

**Features**:
- Database connection pooling
- Automatic result caching
- Metrics and monitoring
- Resource pool management
- Lifecycle hooks (before_start, on_success, on_failure, on_retry, after_return)

**Example**: See `examples/custom-task-classes.md` for comprehensive guide

### 6. Pydantic Validation Template

**Template**: `templates/pydantic-validation.py`

Type-safe tasks with Pydantic model validation.

**Use when**:
- Need strict input validation
- Want type safety
- Complex nested data structures
- API contract enforcement

**Features**:
- Pydantic models for input validation
- Enum types for constrained values
- Custom validators
- Standardized result format
- Comprehensive error messages

### 7. Database Task Template

**Template**: `templates/database-task.py`

Best practices for database operations in tasks.

**Use when**:
- Performing database queries
- Bulk database operations
- Transactional operations
- Database migrations

**Features**:
- Connection pooling
- Parameterized queries (SQL injection prevention)
- Transaction management
- Bulk insert operations
- Pagination support
- Aggregation queries

### 8. API Task Template

**Template**: `templates/api-task.py`

Best practices for external API calls in tasks.

**Use when**:
- Calling external APIs
- Webhook delivery
- Paginated API fetching
- Batch API calls
- API authentication

**Features**:
- Automatic retry on connection errors
- Rate limiting for API quotas
- Timeout handling
- Authentication patterns (Bearer, API key)
- Pagination support
- Batch processing

## Scripts

### generate-task.sh

**Usage**: `./scripts/generate-task.sh <template_name> <output_file> [task_name]`

Generate new Celery task from template with customizations.

**Available templates**:
- basic-task
- retry-task
- rate-limited-task
- time-limited-task
- custom-task-class
- pydantic-validation
- database-task
- api-task

**Example**:
```bash
./scripts/generate-task.sh retry-task tasks/my_api_call.py fetch_data
```

### test-task.sh

**Usage**: `./scripts/test-task.sh <task_file.py> [task_name]`

Test Celery task by:
1. Validating Python syntax
2. Checking imports
3. Verifying task structure
4. Security scanning
5. Best practices check
6. Optionally running task

**Example**:
```bash
./scripts/test-task.sh templates/retry-task.py fetch_api_data
```

### validate-task.sh

**Usage**: `./scripts/validate-task.sh <task_file.py>`

Comprehensive validation including:
- Required elements (Celery import, app initialization, decorators)
- Best practices (docstrings, type hints, error handling, logging)
- Retry configuration
- Security checks (no hardcoded credentials, SQL injection)
- Performance checks (rate limits, time limits)
- Code quality (examples, configuration)
- Pattern detection

**Example**:
```bash
./scripts/validate-task.sh templates/api-task.py
```

## Usage Examples

### Example 1: Create API Task with Retries

```python
# Use retry-task.py template
from templates.retry_task import fetch_api_data

# Queue task
result = fetch_api_data.delay('https://api.example.com/data')

# Get result
data = result.get(timeout=60)
print(data)
```

### Example 2: Rate-Limited Batch Processing

```python
# Use rate-limited-task.py template
from templates.rate_limited_task import api_call_rate_limited

# Process 100 items at 10/minute rate
for i in range(100):
    api_call_rate_limited.delay(f'/endpoint/{i}')

# Celery automatically enforces rate limit
```

### Example 3: Database Operations with Connection Pooling

```python
# Use database-task.py template
from templates.database_task import insert_record, bulk_insert

# Single insert
result1 = insert_record.delay('users', {
    'name': 'John Doe',
    'email': '[email protected]'
})

# Bulk insert
records = [
    {'name': 'Jane', 'email': '[email protected]'},
    {'name': 'Bob', 'email': '[email protected]'},
]
result2 = bulk_insert.delay('users', records)
```

### Example 4: Custom Task with Metrics

```python
# Use custom-task-class.py template
from templates.custom_task_class import monitored_task

# Automatic metrics tracking
result = monitored_task.delay(123)

# Metrics logged automatically:
# - Start time
# - Duration
# - Success/failure
# - Arguments
```

### Example 5: Type-Safe Task with Pydantic

```python
# Use pydantic-validation.py template
from templates.pydantic_validation import process_user

# Valid data
user_data = {
    'user_id': 123,
    'email': '[email protected]',
    'username': 'john_doe',
    'age': 30,
    'tags': ['premium']
}
result = process_user.delay(user_data)

# Invalid data returns structured error
invalid_data = {
    'user_id': -1,  # Invalid
    'email': 'not-email',  # Invalid
}
result = process_user.delay(invalid_data)
# Returns: {'status': 'error', 'errors': [...]}
```

## Best Practices

### 1. Always Use Placeholders for Credentials

```python
# ✅ CORRECT
api_key = os.getenv('API_KEY', 'your_api_key_here')

# ❌ WRONG
api_key = 'sk-abc123xyz456'
```

### 2. Set Appropriate Timeouts

```python
# ✅ CORRECT
response = requests.get(url, timeout=30)

# ❌ WRONG
response = requests.get(url)  # Can hang forever
```

### 3. Use Parameterized Queries

```python
# ✅ CORRECT
db.execute("SELECT * FROM users WHERE id = %s", (user_id,))

# ❌ WRONG
db.execute(f"SELECT * FROM users WHERE id = {user_id}")
```

### 4. Combine Retry with Rate Limiting

```python
@app