Tutorial Background Tasks

Background Tasks with Celery

Run tasks asynchronously without blocking your web requests. Perfect for sending emails, processing uploads, or any slow operation.

Time: ~15 minutes

---

What You'll Build

• Enable Celery with Redis
• Create a background task
• Call it from a route
• Monitor task execution

---

Prerequisites

Redis running locally:

# macOS
brew install redis
brew services start redis

# Ubuntu
sudo apt install redis-server
sudo systemctl start redis

---

Step 1: Install Celery Dependencies

Enferno has optional Celery support:

uv sync --extra full

This installs celery and redis packages.

---

Step 2: Configure Environment

Add to .env:

REDIS_URL=redis://localhost:6379/1
CELERY_BROKER_URL=redis://localhost:6379/2
CELERY_RESULT_BACKEND=redis://localhost:6379/2

---

Step 3: Understand the Existing Setup

Enferno already has Celery configured in enferno/tasks/__init__.py:

if CELERY_AVAILABLE:
    from celery import Celery
    from enferno.settings import Config as cfg

    if cfg.CELERY_BROKER_URL:
        celery = Celery(
            "enferno.tasks",
            broker=cfg.CELERY_BROKER_URL,
            backend=cfg.CELERY_RESULT_BACKEND,
        )

        # Flask app context for tasks
        class ContextTask(celery.Task):
            def __call__(self, *args, **kwargs):
                from enferno.app import create_app
                with create_app(cfg).app_context():
                    return super().__call__(*args, **kwargs)

        celery.Task = ContextTask

The ContextTask ensures your tasks have access to Flask's app context (database, config, etc).

---

Step 4: Create a Task

Prompt for AI assistant:

Add an email notification task to enferno/tasks/init.py that takes an email address and message, logs the send attempt, and simulates sending. Include proper error handling.

Add to enferno/tasks/__init__.py (after the existing code):

if celery:
    @celery.task(bind=True, max_retries=3)
    def send_email_task(self, to_email, subject, body):
        """Send email asynchronously."""
        import time
        from flask import current_app

        try:
            current_app.logger.info(f"Sending email to {to_email}: {subject}")

            # Simulate email sending (replace with real SMTP)
            time.sleep(2)  # Simulates network delay

            current_app.logger.info(f"Email sent to {to_email}")
            return {"status": "sent", "to": to_email}

        except Exception as e:
            current_app.logger.error(f"Email failed: {e}")
            raise self.retry(exc=e, countdown=60)

Pattern notes:

• bind=True gives access to self for retries
• max_retries=3 limits retry attempts
• self.retry(countdown=60) waits 60 seconds before retrying
• Flask app context is available via ContextTask

---

Step 5: Call the Task from a Route

Prompt for AI assistant:

Add a route /api/notify in the portal blueprint that accepts POST with email and message, queues a send_email_task, and returns the task ID.

Add to enferno/portal/views.py:

from enferno.tasks import celery

@portal.route("/api/notify", methods=["POST"])
def send_notification():
    # Check if Celery is available
    if not celery:
        return jsonify({"error": "Background tasks not configured"}), 503

    from enferno.tasks import send_email_task

    data = request.json
    email = data.get("email")
    message = data.get("message", "Hello from Enferno!")

    if not email:
        return jsonify({"error": "Email required"}), 400

    # Queue the task
    task = send_email_task.delay(
        to_email=email,
        subject="Notification",
        body=message
    )

    return jsonify({
        "status": "queued",
        "task_id": task.id
    })

---

Step 6: Run the Worker

Open a second terminal:

cd enferno
source .venv/bin/activate
uv run celery -A enferno.tasks worker -l info

You should see:

[config]
.> app:         enferno.tasks:0x...
.> transport:   redis://localhost:6379/2
...
[queues]
.> celery           exchange=celery(direct) key=celery

celery@hostname ready.

---

Step 7: Test It

With both Flask and Celery running:

curl -X POST http://localhost:5000/dashboard/api/notify \
  -H "Content-Type: application/json" \
  -d '{"email": "test@example.com", "message": "Hello!"}'

Response:

{"status": "queued", "task_id": "abc123..."}

Watch the Celery terminal - you'll see the task execute.

---

Optional: Check Task Status

Add a status endpoint:

@portal.route("/api/task/<task_id>")
def task_status(task_id):
    if not celery:
        return jsonify({"error": "Background tasks not configured"}), 503

    from celery.result import AsyncResult
    result = AsyncResult(task_id, app=celery)

    return jsonify({
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.ready() else None
    })

---

Docker Setup

With Docker Compose, Celery runs automatically. The docker-compose.yml includes:

celery:
  build: .
  command: celery -A enferno.tasks worker -l info
  depends_on:
    - redis
  environment:
    - CELERY_BROKER_URL=redis://:${REDIS_PASSWORD}@redis:6379/2
    - CELERY_RESULT_BACKEND=redis://:${REDIS_PASSWORD}@redis:6379/3

Just run:

docker compose up --build

---

Production with Ignite

Ignite sets up Celery as a systemd service automatically:

# Ignite creates these for you:
/etc/systemd/system/celery-yourdomain.service

View Celery logs:

journalctl -u celery-yourdomain.service -f

---

Common Tasks Examples

Process uploaded file:

@celery.task
def process_upload(file_path):
    # Resize image, generate thumbnails, etc.
    pass

Send bulk emails:

@celery.task
def send_newsletter(user_ids):
    for user_id in user_ids:
        user = db.session.get(User, user_id)
        send_email_task.delay(user.email, "Newsletter", content)

Scheduled cleanup:

# In celery config, add beat schedule
celery.conf.beat_schedule = {
    'cleanup-daily': {
        'task': 'enferno.tasks.cleanup_old_data',
        'schedule': 86400.0,  # 24 hours
    },
}

---

Troubleshooting

Issue	Solution
"Background tasks not configured"	Check CELERY_BROKER_URL in .env
Tasks not executing	Ensure Celery worker is running
Redis connection refused	Start Redis: brew services start redis
Import errors in task	Check task has Flask app context

---

← Back to Tutorials