Image Processing

Build scalable image processing workflows that handle file uploads, generate multiple sizes, and optimize images for web delivery.

Overview

This example demonstrates image processing automation:

Multi-size generation from single uploaded images
Asynchronous processing to avoid blocking web requests
CDN integration for optimized delivery
Error handling for corrupted or invalid files
Progress tracking for long-running operations

Perfect for user-generated content platforms, e-commerce sites, or any application handling image uploads.

Task Definitions

from hyrex import HyrexRegistry
from pydantic import BaseModel
import requests
from PIL import Image
import io
import boto3
from typing import List, Tuple

hy = HyrexRegistry()

class ImageProcessingContext(BaseModel):
    image_url: str
    user_id: int
    sizes: List[Tuple[int, int]] = [(100, 100), (300, 300), (800, 800)]
    quality: int = 85
    output_format: str = "JPEG"

@hy.task(
    queue="image-processing",
    max_retries=3,
    timeout_seconds=300
)
def process_uploaded_image(context: ImageProcessingContext):
    """Generate multiple sizes of an uploaded image"""
    
    try:
        # Download original image
        response = requests.get(context.image_url, timeout=30)
        response.raise_for_status()
        
        # Open with PIL
        original_image = Image.open(io.BytesIO(response.content))
        
        # Convert to RGB if necessary (handles RGBA, etc.)
        if original_image.mode in ('RGBA', 'LA', 'P'):
            original_image = original_image.convert('RGB')
        
        processed_images = []
        s3_client = boto3.client('s3')
        
        for width, height in context.sizes:
            # Resize image maintaining aspect ratio
            resized_image = original_image.copy()
            resized_image.thumbnail((width, height), Image.Resampling.LANCZOS)
            
            # Save to buffer
            buffer = io.BytesIO()
            resized_image.save(
                buffer, 
                format=context.output_format,
                quality=context.quality,
                optimize=True
            )
            buffer.seek(0)
            
            # Upload to S3
            key = f"images/{context.user_id}/{width}x{height}.jpg"
            s3_client.upload_fileobj(
                buffer,
                'your-cdn-bucket',
                key,
                ExtraArgs={'ContentType': 'image/jpeg', 'ACL': 'public-read'}
            )
            
            processed_url = f"https://cdn.example.com/{key}"
            processed_images.append({
                "size": f"{width}x{height}",
                "url": processed_url,
                "width": resized_image.width,
                "height": resized_image.height
            })
        
        return {
            "success": True,
            "original": {
                "url": context.image_url,
                "width": original_image.width,
                "height": original_image.height
            },
            "processed": processed_images,
            "user_id": context.user_id,
            "formats_generated": len(processed_images)
        }
        
    except requests.RequestException as e:
        raise Exception(f"Failed to download image: {str(e)}")
    except Image.UnidentifiedImageError:
        raise Exception("Invalid or corrupted image file")
    except Exception as e:
        raise Exception(f"Image processing failed: {str(e)}")

@hy.task(max_retries=2)
def notify_processing_complete(user_id: int, processing_result: dict):
    """Notify user that image processing is complete"""
    
    if processing_result["success"]:
        # Send success notification
        send_notification(
            user_id,
            "Image processing complete!",
            f"Generated {processing_result['formats_generated']} optimized versions"
        )
    else:
        # Send failure notification
        send_notification(
            user_id,
            "Image processing failed",
            "Please try uploading a different image"
        )
    
    return {"notification_sent": True, "user_id": user_id}

@hy.task
def batch_process_images(image_contexts: List[ImageProcessingContext]):
    """Process multiple images in parallel"""
    
    task_ids = []
    for context in image_contexts:
        task = process_uploaded_image.send(context)
        task_ids.append({
            "task_id": task.task_id,
            "user_id": context.user_id,
            "image_url": context.image_url
        })
    
    return {
        "batch_started": True,
        "total_images": len(image_contexts),
        "task_ids": task_ids
    }

import { HyrexRegistry } from 'hyrex';
import { z } from 'zod';
import fetch from 'node-fetch';
import sharp from 'sharp';
import AWS from 'aws-sdk';

const hy = new HyrexRegistry();
const s3 = new AWS.S3();

const ImageProcessingSchema = z.object({
    imageUrl: z.string().url(),
    userId: z.number(),
    sizes: z.array(z.tuple([z.number(), z.number()]))
        .default([[100, 100], [300, 300], [800, 800]]),
    quality: z.number().default(85),
    outputFormat: z.string().default("jpeg")
});

type ImageProcessingContext = z.infer<typeof ImageProcessingSchema>;

const processUploadedImage = hy.task({
    name: 'processUploadedImage',
    config: {
        queue: 'image-processing',
        maxRetries: 3,
        timeoutSeconds: 300
    },
    argSchema: ImageProcessingSchema,
    func: async (context: ImageProcessingContext) => {
        try {
            // Download original image
            const response = await fetch(context.imageUrl, { timeout: 30000 });
            if (!response.ok) {
                throw new Error(`HTTP ${response.status}: ${response.statusText}`);
            }
            
            const imageBuffer = await response.buffer();
            
            // Get original image metadata
            const originalInfo = await sharp(imageBuffer).metadata();
            
            const processedImages = [];
            
            for (const [width, height] of context.sizes) {
                // Resize image maintaining aspect ratio
                const resizedBuffer = await sharp(imageBuffer)
                    .resize(width, height, {
                        fit: 'inside',
                        withoutEnlargement: true
                    })
                    .jpeg({ quality: context.quality })
                    .toBuffer();
                
                // Get resized image info
                const resizedInfo = await sharp(resizedBuffer).metadata();
                
                // Upload to S3
                const key = `images/${context.userId}/${width}x${height}.jpg`;
                await s3.upload({
                    Bucket: 'your-cdn-bucket',
                    Key: key,
                    Body: resizedBuffer,
                    ContentType: 'image/jpeg',
                    ACL: 'public-read'
                }).promise();
                
                const processedUrl = `https://cdn.example.com/${key}`;
                processedImages.push({
                    size: `${width}x${height}`,
                    url: processedUrl,
                    width: resizedInfo.width || 0,
                    height: resizedInfo.height || 0
                });
            }
            
            return {
                success: true,
                original: {
                    url: context.imageUrl,
                    width: originalInfo.width || 0,
                    height: originalInfo.height || 0
                },
                processed: processedImages,
                userId: context.userId,
                formatsGenerated: processedImages.length
            };
            
        } catch (error: any) {
            if (error.message.includes('HTTP')) {
                throw new Error(`Failed to download image: ${error.message}`);
            } else if (error.message.includes('Input buffer')) {
                throw new Error("Invalid or corrupted image file");
            } else {
                throw new Error(`Image processing failed: ${error.message}`);
            }
        }
    }
});

const notifyProcessingComplete = hy.task({
    name: 'notifyProcessingComplete',
    config: { maxRetries: 2 },
    func: async (userId: number, processingResult: any) => {
        if (processingResult.success) {
            // Send success notification
            await sendNotification(
                userId,
                "Image processing complete!",
                `Generated ${processingResult.formatsGenerated} optimized versions`
            );
        } else {
            // Send failure notification
            await sendNotification(
                userId,
                "Image processing failed",
                "Please try uploading a different image"
            );
        }
        
        return { notificationSent: true, userId };
    }
});

const batchProcessImages = hy.task({
    name: 'batchProcessImages',
    func: async (imageContexts: ImageProcessingContext[]) => {
        const taskIds = [];
        
        for (const context of imageContexts) {
            const task = await processUploadedImage.send(context);
            taskIds.push({
                taskId: task.taskId,
                userId: context.userId,
                imageUrl: context.imageUrl
            });
        }
        
        return {
            batchStarted: true,
            totalImages: imageContexts.length,
            taskIds
        };
    }
});

REST API Integration

from fastapi import FastAPI, File, UploadFile, HTTPException
from pydantic import BaseModel
import boto3
import uuid
from .tasks import process_uploaded_image, batch_process_images

app = FastAPI()
s3_client = boto3.client('s3')

class ProcessImageRequest(BaseModel):
    image_url: str
    user_id: int
    sizes: list[tuple[int, int]] = [(100, 100), (300, 300), (800, 800)]

@app.post("/images/upload")
async def upload_and_process_image(file: UploadFile = File(...), user_id: int = 1):
    """Upload image and trigger processing"""
    
    # Validate file type
    if not file.content_type or not file.content_type.startswith('image/'):
        raise HTTPException(status_code=400, detail="File must be an image")
    
    try:
        # Upload original to S3
        file_extension = file.filename.split('.')[-1] if '.' in file.filename else 'jpg'
        key = f"uploads/{user_id}/{uuid.uuid4()}.{file_extension}"
        
        s3_client.upload_fileobj(
            file.file,
            'your-upload-bucket',
            key,
            ExtraArgs={'ContentType': file.content_type}
        )
        
        image_url = f"https://your-upload-bucket.s3.amazonaws.com/{key}"
        
        # Queue processing task
        task = process_uploaded_image.send(ImageProcessingContext(
            image_url=image_url,
            user_id=user_id
        ))
        
        return {
            "message": "Image uploaded and processing started",
            "task_id": task.task_id,
            "original_url": image_url,
            "user_id": user_id
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")

@app.post("/images/process")
async def process_existing_image(request: ProcessImageRequest):
    """Process an existing image by URL"""
    
    task = process_uploaded_image.send(ImageProcessingContext(**request.dict()))
    
    return {
        "message": "Image processing started",
        "task_id": task.task_id,
        "image_url": request.image_url
    }

@app.post("/images/batch")
async def batch_process_images_endpoint(requests: list[ProcessImageRequest]):
    """Process multiple images in batch"""
    
    contexts = [ImageProcessingContext(**req.dict()) for req in requests]
    task = batch_process_images.send(contexts)
    
    return {
        "message": f"Batch processing started for {len(contexts)} images",
        "task_id": task.task_id,
        "total_images": len(contexts)
    }

@app.get("/images/status/{task_id}")
async def get_processing_status(task_id: str):
    """Get image processing status"""
    
    task = hy.get_task(task_id)
    
    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.is_complete else None,
        "error": task.error if task.has_failed else None
    }

import express from 'express';
import multer from 'multer';
import AWS from 'aws-sdk';
import { v4 as uuidv4 } from 'uuid';
import { processUploadedImage, batchProcessImages } from './tasks';

const app = express();
const s3 = new AWS.S3();
const upload = multer({ storage: multer.memoryStorage() });

interface ProcessImageRequest {
    imageUrl: string;
    userId: number;
    sizes?: [number, number][];
}

app.post('/images/upload', upload.single('file'), async (req, res) => {
    if (!req.file) {
        return res.status(400).json({ error: 'No file uploaded' });
    }
    
    if (!req.file.mimetype?.startsWith('image/')) {
        return res.status(400).json({ error: 'File must be an image' });
    }
    
    try {
        const userId = parseInt(req.body.userId) || 1;
        const fileExtension = req.file.originalname?.split('.').pop() || 'jpg';
        const key = `uploads/${userId}/${uuidv4()}.${fileExtension}`;
        
        // Upload original to S3
        await s3.upload({
            Bucket: 'your-upload-bucket',
            Key: key,
            Body: req.file.buffer,
            ContentType: req.file.mimetype
        }).promise();
        
        const imageUrl = `https://your-upload-bucket.s3.amazonaws.com/${key}`;
        
        // Queue processing task
        const task = await processUploadedImage.send({
            imageUrl,
            userId
        });
        
        res.json({
            message: 'Image uploaded and processing started',
            taskId: task.taskId,
            originalUrl: imageUrl,
            userId
        });
        
    } catch (error: any) {
        res.status(500).json({ error: `Upload failed: ${error.message}` });
    }
});

app.post('/images/process', async (req, res) => {
    const { imageUrl, userId, sizes }: ProcessImageRequest = req.body;
    
    try {
        const task = await processUploadedImage.send({
            imageUrl,
            userId,
            sizes
        });
        
        res.json({
            message: 'Image processing started',
            taskId: task.taskId,
            imageUrl
        });
    } catch (error: any) {
        res.status(500).json({ error: error.message });
    }
});

app.post('/images/batch', async (req, res) => {
    const requests: ProcessImageRequest[] = req.body;
    
    try {
        const contexts = requests.map(req => ({
            imageUrl: req.imageUrl,
            userId: req.userId,
            sizes: req.sizes
        }));
        
        const task = await batchProcessImages.send(contexts);
        
        res.json({
            message: `Batch processing started for ${contexts.length} images`,
            taskId: task.taskId,
            totalImages: contexts.length
        });
    } catch (error: any) {
        res.status(500).json({ error: error.message });
    }
});

app.get('/images/status/:taskId', async (req, res) => {
    const { taskId } = req.params;
    
    try {
        const task = await hy.getTask(taskId);
        
        res.json({
            taskId,
            status: task.status,
            result: task.isComplete ? task.result : null,
            error: task.hasFailed ? task.error : null
        });
    } catch (error: any) {
        res.status(404).json({ error: 'Task not found' });
    }
});

Usage Examples

# Upload and process image
curl -X POST http://localhost:8000/images/upload \
  -F "file=@photo.jpg" \
  -F "userId=123"

# Process existing image by URL
curl -X POST http://localhost:8000/images/process \
  -H "Content-Type: application/json" \
  -d '{
    "image_url": "https://example.com/original.jpg",
    "user_id": 123,
    "sizes": [[150, 150], [500, 500]]
  }'

# Check processing status
curl http://localhost:8000/images/status/task_12345

Dependencies

Python Requirements

Pillow==10.0.0
boto3==1.29.0
requests==2.31.0

TypeScript Dependencies

{
  "dependencies": {
    "sharp": "^0.32.0",
    "aws-sdk": "^2.1400.0",
    "node-fetch": "^3.3.0"
  }
}

Production Considerations

File validation: Validate image types and sizes before processing
Storage optimization: Use appropriate S3 storage classes for different image sizes
CDN configuration: Set up CloudFront for optimal image delivery
Error handling: Implement fallbacks for corrupted or invalid images
Memory management: Monitor memory usage for large image processing
Cost optimization: Consider image processing costs and optimize accordingly

Image Processing

Overview

Task Definitions

REST API Integration

Usage Examples

Dependencies

Python Requirements

TypeScript Dependencies

Production Considerations

Next Steps

User Onboarding

Maintenance Tasks

​Overview

​Task Definitions

​REST API Integration

​Usage Examples

​Dependencies

​Python Requirements

​TypeScript Dependencies

​Production Considerations

​Next Steps

User Onboarding

Maintenance Tasks

Overview

Task Definitions

REST API Integration

Usage Examples

Dependencies

Python Requirements

TypeScript Dependencies

Production Considerations

Next Steps