adelabdelgawad 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
# 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
# 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
# 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
# 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
async def trigger_job(
self,
session: AsyncSession,
job_id: str
) -> JobExecution:
"""Manually trigger job execution."""
job = await self._repo.get_job(session, job_id)
execution = await self._repo.create_execution(session, job_id)
# Dispatch to Celery or run inline
from tasks.celery_bridge import dispatch_to_celery
task_id = dispatch_to_celery(
job.job_key,
execution_id=execution.id
)
if not task_id:
# Run inline
await self._execute_job(session, job, execution)
return execution
5. Database Models
# db/models.py
class ScheduledJob(Base):
__tablename__ = "scheduled_jobs"
id = Column(String(36), primary_key=True)
job_key = Column(String(100), unique=True, nullable=False)
name_en = Column(String(200))
name_ar = Column(String(200))
job_type = Column(String(20)) # "interval" or "cron"
# Interval config
interval_seconds = Column(Integer)
interval_minutes = Column(Integer)
interval_hours = Column(Integer)
interval_days = Column(Integer)
# Cron config
cron_expression = Column(String(100))
# APScheduler config
max_instances = Column(Integer, default=1)
misfire_grace_time = Column(Integer, default=60)
coalesce = Column(Boolean, default=True)
is_enabled = Column(Boolean, default=True)
is_active = Column(Boolean, default=True)
executions = relationship("ScheduledJobExecution", back_populates="job")
class ScheduledJobExecution(Base):
__tablename__ = "scheduled_job_executions"
id = Column(String(36), primary_key=True)
job_id = Column(String(36), ForeignKey("scheduled_jobs.id"))
execution_id = Column(String(36)) # Unique per execution
scheduled_at = Column(DateTime(timezone=True))
started_at = Column(DateTime(timezone=True))
completed_at = Column(DateTime(timezone=True))
duration_ms = Column(Integer)
status = Column(String(20)) # pending, running, success, failed
error_message = Column(Text)
result_summary = Column(Text)
job = relationship("ScheduledJob", back_populates="executions")
Key Patterns
Job Types
- Interval: Run every N seconds/minutes/hours/days
- Cron: Run on cron schedule (e.g., "0 0 * * *")
Execution Modes
- Celery: Distributed across workers (CELERY_ENABLED=True)
- Inline: Direct execution in FastAPI (CELERY_ENABLED=False)
Retry Strategy
- Exponential backoff with jitter
- Configurable max retries
- Task-level time limits
Instance Coordination
- Distributed locking for job execution
- Heartbeat tracking for instance health
- Automatic cleanup of stale instances
References
See the references/ directory for:
celery-pattern.md- Celery configuration and tasksscheduler-pattern.md- APScheduler integrationtask-pattern.md- Task definition patternsapi-pattern.md- Scheduler API endpoints