tasks-management

What this skill does

# Tasks Management Skill

Background task management system using APScheduler for scheduling and Celery for distributed execution.

## When to Use This Skill

Use this skill when asked to:
- Set up background task processing
- Create scheduled jobs with APScheduler
- Implement Celery workers for task execution
- Add job management endpoints
- Configure task queues and retries

## Architecture Overview

```
┌─────────────────────────────────────────────────────────────┐
│                    FastAPI Application                       │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │  Scheduler  │  │   Tasks     │  │    API      │         │
│  │   Router    │  │   Bridge    │  │  Endpoints  │         │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘         │
└─────────┼────────────────┼────────────────┼─────────────────┘
          │                │                │
          ▼                ▼                ▼
┌─────────────────────────────────────────────────────────────┐
│                  Scheduler Service                           │
│  • Job CRUD operations                                       │
│  • Execution tracking                                        │
│  • Distributed locking                                       │
│  • Instance management                                       │
└─────────────────────────────┬───────────────────────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        │                     │                     │
        ▼                     ▼                     ▼
┌───────────────┐   ┌─────────────────┐   ┌───────────────┐
│  APScheduler  │   │  Celery Bridge  │   │   Database    │
│  (Scheduling) │   │  (Dispatch)     │   │  (Persistence)│
└───────────────┘   └────────┬────────┘   └───────────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │  Redis Broker   │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │  Celery Worker  │
                    │  (Execution)    │
                    └─────────────────┘
```

## Directory Structure

```
project/
├── celery_app.py              # Celery configuration
├── tasks/
│   ├── __init__.py            # Task exports
│   ├── celery_bridge.py       # APScheduler-Celery bridge
│   ├── email.py               # Email tasks
│   ├── attendance.py          # Domain-specific tasks
│   ├── hris.py                # Integration tasks
│   └── scheduler.py           # Scheduler maintenance
├── api/
│   ├── v1/
│   │   └── scheduler.py       # Scheduler endpoints
│   ├── services/
│   │   └── scheduler_service.py
│   ├── repositories/
│   │   └── scheduler_repository.py
│   └── schemas/
│       └── scheduler_schema.py
└── db/
    └── models.py              # Job & execution models
```

## Core Components

### 1. Celery App Configuration

```python
# celery_app.py
from celery import Celery
from settings import settings

celery_app = Celery(
    "app_name",
    broker=settings.CELERY_BROKER_URL,
    backend=settings.CELERY_RESULT_BACKEND,
    include=[
        "tasks.email",
        "tasks.attendance",
        "tasks.scheduler",
    ],
)

celery_app.conf.update(
    # Serialization
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    
    # Reliability
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    task_track_started=True,
    
    # Result expiration
    result_expires=86400,
    
    # Worker settings
    worker_prefetch_multiplier=1,
    worker_concurrency=10,
    
    # Time limits
    task_soft_time_limit=300,
    task_time_limit=360,
    
    # Retry defaults
    task_default_retry_delay=60,
    
    # Timezone
    timezone="UTC",
    enable_utc=True,
)
```

### 2. Celery-APScheduler Bridge

```python
# tasks/celery_bridge.py
from typing import Optional
from settings import settings

_CELERY_TASK_REGISTRY = {}

def register_celery_task(job_key: str, task):
    """Register a Celery task for a job key."""
    _CELERY_TASK_REGISTRY[job_key] = task

def dispatch_to_celery(
    job_key: str, 
    execution_id: Optional[str] = None, 
    **kwargs
) -> Optional[str]:
    """Dispatch job to Celery if enabled."""
    if not settings.CELERY_ENABLED:
        return None  # Fall back to inline
    
    task = _CELERY_TASK_REGISTRY.get(job_key)
    if not task:
        return None  # Fall back to inline
    
    result = task.delay(execution_id=execution_id, **kwargs)
    return result.id

def initialize_celery_tasks():
    """Register all Celery tasks at startup."""
    from tasks.attendance import sync_attendance_task
    from tasks.scheduler import cleanup_history_task
    
    register_celery_task("attendance_sync", sync_attendance_task)
    register_celery_task("history_cleanup", cleanup_history_task)
```

### 3. Task Definition Pattern

```python
# tasks/scheduler.py
import asyncio
from celery import shared_task

def _run_async(coro):
    """Run async code in Celery worker."""
    try:
        loop = asyncio.get_running_loop()
        import concurrent.futures
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            return executor.submit(asyncio.run, coro).result()
    except RuntimeError:
        return asyncio.run(coro)

@shared_task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_backoff_max=300,
    retry_jitter=True,
    soft_time_limit=120,
    time_limit=180,
)
def cleanup_history_task(
    self,
    execution_id: str = None,
    retention_days: int = 30,
) -> dict:
    """Celery task with automatic retries."""
    
    async def _execute():
        from db.database import get_session
        from api.services.scheduler_service import get_scheduler_service
        
        async with get_session() as session:
            service = get_scheduler_service()
            result = await service.cleanup_history(session, retention_days)
            
            # Update execution status
            if execution_id:
                await update_execution_status(session, execution_id, "success")
            
            return result
    
    return _run_async(_execute())
```

### 4. Scheduler Service

```python
# api/services/scheduler_service.py
class SchedulerService:
    """Manages APScheduler jobs with database persistence."""
    
    _instance: Optional["SchedulerService"] = None
    _scheduler: Optional[AsyncIOScheduler] = None
    
    def __init__(self):
        self._repo = SchedulerRepository()
        self._job_functions: Dict[str, Callable] = {}
    
    @classmethod
    def get_instance(cls) -> "SchedulerService":
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
    
    async def initialize(self, session: AsyncSession, mode: str = "embedded"):
        """Initialize scheduler and register instance."""
        instance = await self._repo.register_instance(session, mode)
        self._scheduler = AsyncIOScheduler()
        return instance.id
    
    async def start(self, session: AsyncSession):
        """Start scheduler and load jobs from database."""
        if settings.CELERY_ENABLED:
            from tasks.celery_bridge import initialize_celery_tasks
            initialize_celery_tasks()
        
        jobs = await self._repo.get_enabled_jobs(session)
        for job in jobs:
            await self._add_job_to_scheduler(job)
        
        self._scheduler.start()
    
    async def create_job(
        self, 
        session: AsyncSession, 
        job_data: ScheduledJobCreate
    ) -> ScheduledJob:
        """Create new scheduled job."""
        job = await self._repo.create_job(session, job_data)
        if job.is_enabled:
            await self._add_job_to_scheduler(job)
        return job