📘 Background Tasks: Celery with Flask/Django

🎯 Introduction

Welcome to this exciting tutorial on background tasks with Celery! 🎉 In this guide, we’ll explore how to handle time-consuming operations without making your users wait.

Ever wondered how Netflix processes your video uploads 🎬, or how your favorite e-commerce site sends order confirmation emails instantly? The secret is background tasks! You’ll discover how Celery can transform your Flask and Django applications into powerhouses that handle heavy work behind the scenes.

By the end of this tutorial, you’ll feel confident implementing background tasks in your own projects! Let’s dive in! 🏊‍♂️

📚 Understanding Background Tasks with Celery

🤔 What are Background Tasks?

Background tasks are like having a personal assistant 🤝. Think of it as delegating time-consuming work to someone else while you continue serving your customers. Your web app stays responsive while the heavy lifting happens behind the scenes!

In Python terms, background tasks allow you to offload slow operations to separate worker processes. This means you can:

• ✨ Keep your web app lightning fast
• 🚀 Process multiple tasks simultaneously
• 🛡️ Handle failures gracefully with retries

💡 Why Use Celery?

Here’s why developers love Celery:

1. Battle-tested 🔒: Used by Instagram, Mozilla, and more
2. Flexible 💻: Works with Flask, Django, and plain Python
3. Reliable 📖: Built-in retries and error handling
4. Scalable 🔧: Add more workers as you grow

Real-world example: Imagine building an online store 🛒. With Celery, you can process orders instantly while emails, inventory updates, and analytics happen in the background!

🔧 Basic Syntax and Usage

📝 Setting Up Celery

Let’s start with a friendly example:

# 👋 Hello, Celery!
from celery import Celery

# 🎨 Create our Celery app
app = Celery('tasks', broker='redis://localhost:6379')

# 🎯 Define a simple task
@app.task
def send_welcome_email(user_email):
    # 📧 Simulate sending email
    print(f"Sending welcome email to {user_email}! 🎉")
    return f"Email sent to {user_email}"

💡 Explanation: The @app.task decorator turns any function into a background task. Redis acts as our message broker - think of it as a todo list for our workers!

🎯 Common Patterns

Here are patterns you’ll use daily:

# 🏗️ Pattern 1: Flask Integration
from flask import Flask, jsonify
from celery import Celery

flask_app = Flask(__name__)
celery_app = Celery(flask_app.name, broker='redis://localhost:6379')

# 🎨 Pattern 2: Task with retry logic
@celery_app.task(bind=True, max_retries=3)
def process_image(self, image_path):
    try:
        # 🖼️ Process the image
        result = resize_image(image_path)
        return result
    except Exception as exc:
        # 🔄 Retry if something goes wrong
        raise self.retry(exc=exc, countdown=60)

# 🔄 Pattern 3: Scheduled tasks
from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    'daily-report': {
        'task': 'tasks.generate_report',
        'schedule': crontab(hour=9, minute=0),  # 📅 Every day at 9 AM
    },
}

💡 Practical Examples

🛒 Example 1: E-commerce Order Processing

Let’s build something real:

# 🛍️ Order processing system
from celery import Celery, group
from datetime import datetime

app = Celery('shop', broker='redis://localhost:6379')

# 📦 Order processing task
@app.task
def process_order(order_id, customer_email, items):
    print(f"🛒 Processing order #{order_id}")
    
    # 💰 Calculate total
    total = sum(item['price'] * item['quantity'] for item in items)
    
    # 🎯 Chain multiple tasks
    workflow = group(
        update_inventory.s(items),
        send_order_email.s(customer_email, order_id),
        notify_warehouse.s(order_id),
        update_analytics.s(order_id, total)
    )
    
    workflow.apply_async()
    return f"Order #{order_id} processed! Total: ${total}"

# 📊 Inventory update
@app.task
def update_inventory(items):
    for item in items:
        print(f"📉 Reducing stock for {item['name']} by {item['quantity']}")
    return "Inventory updated! ✅"

# 📧 Email notification
@app.task
def send_order_email(customer_email, order_id):
    print(f"📧 Sending confirmation to {customer_email}")
    # Simulate email sending
    return f"Email sent for order #{order_id}! 💌"

# 🏭 Warehouse notification
@app.task
def notify_warehouse(order_id):
    print(f"🚚 Notifying warehouse about order #{order_id}")
    return "Warehouse notified! 📦"

# 📈 Analytics update
@app.task
def update_analytics(order_id, total):
    print(f"📊 Recording sale: Order #{order_id}, Amount: ${total}")
    return "Analytics updated! 📈"

# 🎮 Let's use it!
if __name__ == '__main__':
    items = [
        {'name': 'Python Book', 'price': 29.99, 'quantity': 2, 'emoji': '📘'},
        {'name': 'Coffee Mug', 'price': 12.99, 'quantity': 1, 'emoji': '☕'}
    ]
    
    result = process_order.delay(
        order_id='ORD-001',
        customer_email='[email protected]',
        items=items
    )
    print(f"🎉 Task ID: {result.id}")

🎯 Try it yourself: Add a discount calculation task and priority shipping notification!

🎮 Example 2: Django Social Media Post Processor

Let’s make it fun with Django:

# 🏆 Social media post processor
from celery import shared_task
from django.core.cache import cache
from PIL import Image
import requests

# 📸 Image processing task
@shared_task(bind=True)
def process_social_post(self, post_id, image_url, caption):
    print(f"🎨 Processing post #{post_id}")
    
    try:
        # 📥 Download image
        response = requests.get(image_url)
        img = Image.open(BytesIO(response.content))
        
        # 🎯 Create multiple versions
        versions = {
            'thumbnail': (150, 150),
            'mobile': (480, 480),
            'desktop': (1080, 1080)
        }
        
        processed_urls = {}
        for version_name, size in versions.items():
            # 🖼️ Resize image
            resized = img.resize(size, Image.LANCZOS)
            url = save_to_storage(resized, f"{post_id}_{version_name}")
            processed_urls[version_name] = url
            print(f"✨ Created {version_name} version!")
        
        # 🏷️ Extract hashtags
        hashtags = extract_hashtags.delay(caption)
        
        # 😊 Analyze sentiment
        sentiment = analyze_sentiment.delay(caption)
        
        # 💾 Cache results
        cache.set(f"post_{post_id}_urls", processed_urls, 3600)
        
        return {
            'status': 'success',
            'urls': processed_urls,
            'message': f'Post #{post_id} processed! 🎉'
        }
        
    except Exception as exc:
        # 🔄 Retry on failure
        print(f"⚠️ Error processing post: {exc}")
        raise self.retry(exc=exc, countdown=30, max_retries=3)

# 🏷️ Hashtag extraction
@shared_task
def extract_hashtags(caption):
    import re
    hashtags = re.findall(r'#\w+', caption)
    print(f"🏷️ Found hashtags: {hashtags}")
    return hashtags

# 😊 Sentiment analysis
@shared_task
def analyze_sentiment(caption):
    # Simulate sentiment analysis
    if any(word in caption.lower() for word in ['love', 'amazing', 'great']):
        sentiment = 'positive 😊'
    elif any(word in caption.lower() for word in ['hate', 'terrible', 'bad']):
        sentiment = 'negative 😢'
    else:
        sentiment = 'neutral 😐'
    
    print(f"💭 Sentiment: {sentiment}")
    return sentiment

# 🎯 Django view example
from django.http import JsonResponse
from django.views import View

class CreatePostView(View):
    def post(self, request):
        # 📝 Get post data
        image_url = request.POST.get('image_url')
        caption = request.POST.get('caption')
        
        # 🚀 Queue for processing
        task = process_social_post.delay(
            post_id=generate_post_id(),
            image_url=image_url,
            caption=caption
        )
        
        return JsonResponse({
            'message': 'Post queued for processing! 🎉',
            'task_id': task.id,
            'status_url': f'/api/task-status/{task.id}/'
        })

🚀 Advanced Concepts

🧙‍♂️ Advanced Topic 1: Task Workflows

When you’re ready to level up, try task chains and groups:

# 🎯 Advanced workflow patterns
from celery import chain, group, chord

# 🔗 Chain tasks (run sequentially)
workflow = chain(
    fetch_data.s(user_id),
    process_data.s(),
    generate_report.s(),
    send_report.s()
)

# 🎨 Group tasks (run in parallel)
parallel_tasks = group(
    resize_image.s(img_id, 'small'),
    resize_image.s(img_id, 'medium'),
    resize_image.s(img_id, 'large')
)

# 🎵 Chord (parallel then callback)
@app.task
def aggregate_results(results):
    return f"Processed {len(results)} items! ✨"

chord_workflow = chord(
    (process_item.s(item) for item in items),
    aggregate_results.s()
)

🏗️ Advanced Topic 2: Monitoring and Rate Limiting

For production-ready apps:

# 🚀 Rate limiting tasks
@app.task(rate_limit='10/m')  # 10 per minute
def send_notification(user_id, message):
    print(f"📱 Sending notification to user {user_id}")
    return "Notification sent! 💬"

# 📊 Task monitoring
@app.task(bind=True)
def monitored_task(self, data):
    # 📈 Track progress
    total = len(data)
    for i, item in enumerate(data):
        process_item(item)
        # 🎯 Update task progress
        self.update_state(
            state='PROGRESS',
            meta={'current': i, 'total': total}
        )
    
    return f"Processed {total} items! 🎉"

# 🛡️ Error handling with custom states
@app.task(bind=True)
def robust_task(self, risky_data):
    try:
        result = risky_operation(risky_data)
        return result
    except ValidationError as exc:
        # ⚠️ Custom error state
        self.update_state(
            state='VALIDATION_ERROR',
            meta={'error': str(exc)}
        )
        raise

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Memory Leaks with Large Data

# ❌ Wrong way - passing large objects!
@app.task
def process_huge_data(massive_dataframe):
    # 💥 This will consume lots of memory!
    return massive_dataframe.mean()

# ✅ Correct way - pass references!
@app.task
def process_huge_data(file_path):
    # 📁 Load data inside the task
    df = pd.read_csv(file_path)
    result = df.mean()
    return result

🤯 Pitfall 2: Task Result Backend Overflow

# ❌ Dangerous - storing large results!
@app.task
def generate_report():
    huge_report = create_massive_report()  # 💥 100MB report!
    return huge_report

# ✅ Safe - store reference instead!
@app.task
def generate_report():
    report_path = create_and_save_report()  # 💾 Save to storage
    return {
        'status': 'complete',
        'report_url': f'/reports/{report_path}',
        'size_mb': get_file_size(report_path)
    }

🛠️ Best Practices

1. 🎯 Keep Tasks Atomic: Each task should do one thing well
2. 📝 Use Task Names: Give tasks descriptive names for monitoring
3. 🛡️ Set Time Limits: Prevent tasks from running forever
4. 🎨 Use Result Backends Wisely: Don’t store large results
5. ✨ Handle Failures Gracefully: Always plan for retries

🧪 Hands-On Exercise

🎯 Challenge: Build a Video Processing Pipeline

Create a background task system for video processing:

📋 Requirements:

• ✅ Accept video upload and queue for processing
• 🏷️ Extract video metadata (duration, resolution)
• 👤 Generate multiple quality versions (480p, 720p, 1080p)
• 📅 Create thumbnail at specific timestamps
• 🎨 Apply watermark to all versions!

🚀 Bonus Points:

• Add progress tracking for long videos
• Implement priority queue for premium users
• Send email notification when complete

💡 Solution

# 🎯 Video processing pipeline!
from celery import Celery, chain, group
from moviepy.editor import VideoFileClip
import os

app = Celery('video_processor', broker='redis://localhost:6379')

# 📹 Main video processing task
@app.task(bind=True)
def process_video(self, video_id, video_path, user_email):
    print(f"🎬 Processing video {video_id}")
    
    # 📊 Extract metadata
    metadata = extract_metadata.delay(video_path)
    
    # 🎯 Create processing workflow
    workflow = chain(
        validate_video.s(video_path),
        group(
            create_quality_version.s(video_path, '480p'),
            create_quality_version.s(video_path, '720p'),
            create_quality_version.s(video_path, '1080p'),
        ),
        generate_thumbnails.s(video_path),
        apply_watermark.s(video_id),
        notify_completion.s(user_email, video_id)
    )
    
    return workflow.apply_async()

# 🔍 Metadata extraction
@app.task
def extract_metadata(video_path):
    clip = VideoFileClip(video_path)
    metadata = {
        'duration': clip.duration,
        'fps': clip.fps,
        'resolution': f"{clip.w}x{clip.h}",
        'size_mb': os.path.getsize(video_path) / (1024 * 1024)
    }
    print(f"📊 Metadata: {metadata}")
    return metadata

# ✅ Video validation
@app.task
def validate_video(video_path):
    try:
        clip = VideoFileClip(video_path)
        if clip.duration > 3600:  # 1 hour limit
            raise ValueError("Video too long! Max 1 hour 🕐")
        return True
    except Exception as e:
        print(f"❌ Validation failed: {e}")
        raise

# 🎬 Quality version creation
@app.task(bind=True)
def create_quality_version(self, video_path, quality):
    resolutions = {
        '480p': (854, 480),
        '720p': (1280, 720),
        '1080p': (1920, 1080)
    }
    
    try:
        clip = VideoFileClip(video_path)
        width, height = resolutions[quality]
        
        # 📐 Resize video
        resized = clip.resize(height=height)
        output_path = f"{video_path}_{quality}.mp4"
        
        # 💾 Save with progress tracking
        resized.write_videofile(
            output_path,
            codec='libx264',
            audio_codec='aac',
            progress_bar=False,
            logger=None
        )
        
        print(f"✨ Created {quality} version!")
        return output_path
        
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

# 🖼️ Thumbnail generation
@app.task
def generate_thumbnails(video_path):
    clip = VideoFileClip(video_path)
    timestamps = [0, clip.duration/4, clip.duration/2, clip.duration*3/4]
    
    thumbnails = []
    for i, timestamp in enumerate(timestamps):
        thumbnail_path = f"{video_path}_thumb_{i}.jpg"
        clip.save_frame(thumbnail_path, t=timestamp)
        thumbnails.append(thumbnail_path)
        print(f"📸 Generated thumbnail {i+1}/4")
    
    return thumbnails

# 💧 Watermark application
@app.task
def apply_watermark(video_versions, video_id):
    # Simulate watermarking
    print(f"💧 Applying watermark to all versions")
    return "Watermark applied! 🎨"

# 📧 Completion notification
@app.task
def notify_completion(watermark_result, user_email, video_id):
    print(f"📧 Sending completion email to {user_email}")
    return f"Video {video_id} processing complete! 🎉"

# 🎮 Usage example
if __name__ == '__main__':
    task = process_video.delay(
        video_id='VID-001',
        video_path='/uploads/awesome_video.mp4',
        user_email='[email protected]'
    )
    print(f"🚀 Processing started! Task ID: {task.id}")

🎓 Key Takeaways

You’ve learned so much! Here’s what you can now do:

• ✅ Create background tasks with Celery confidence 💪
• ✅ Integrate with Flask/Django seamlessly 🛡️
• ✅ Build task workflows for complex operations 🎯
• ✅ Handle failures gracefully with retries 🐛
• ✅ Scale your applications with worker processes! 🚀

Remember: Background tasks are your secret weapon for building responsive, scalable applications! 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered background tasks with Celery!

Here’s what to do next:

1. 💻 Practice with the video processing exercise
2. 🏗️ Add Celery to your existing Flask/Django project
3. 📚 Explore Celery Beat for scheduled tasks
4. 🌟 Learn about task monitoring with Flower!

Remember: Every scalable app started with its first background task. Keep coding, keep learning, and most importantly, have fun! 🚀

Happy coding! 🎉🚀✨