celery-tasks

Celery Tasks

context7: If the mcp_context7 tool is available, resolve and load the full celery documentation before making any changes to the task system:

mcp_context7_resolve-library-id: "celery"
mcp_context7_get-library-docs: <resolved-id>

The Celery application is defined in {{cookiecutter.__package_slug}}/celery.py. Tasks are exposed via the @celery.task decorator.

---

Defining Tasks

Import the celery app instance and decorate functions:

from logging import getLogger
from {{cookiecutter.__package_slug}}.celery import celery

logger = getLogger(__name__)


@celery.task
def send_email(to: str, subject: str, body: str) -> dict[str, str]:
    """Send an email asynchronously."""
    logger.info(f"Sending email to {to}: {subject}")
    return {"status": "sent", "to": to}

Rules:

• Use logger (never print) for all output
• Pass IDs, not objects — tasks serialize arguments, complex objects can't be serialized reliably
• Return simple types (dict, list, primitives) — not ORM instances

---

Task Organization

Organize tasks in separate modules under {{cookiecutter.__package_slug}}/tasks/:

{{cookiecutter.__package_slug}}/
├── celery.py           # Celery app configuration
└── tasks/
    ├── __init__.py
    ├── email.py        # Email-related tasks
    └── reports.py      # Report generation tasks

Import task modules in {{cookiecutter.__package_slug}}/celery.py to ensure registration:

from {{cookiecutter.__package_slug}}.tasks import email, reports

---

Calling Tasks

# Fire and forget
send_email.delay("[email protected]", "Welcome", "Thanks for signing up!")

# With options
send_email.apply_async(
    args=["[email protected]", "Welcome", "Body"],
    countdown=60,       # Execute after 60 seconds
    queue='emails',     # Route to specific queue
)

# Get result (blocking)
result = send_email.delay("[email protected]", "Hello", "Body")
output = result.get(timeout=10)