Simple Batch Operations

---

This guide shows how to test batch/bulk API endpoints that process multiple items at once. Perfect for testing batch create, update, and delete operations.

Use Cases

• Test batch create operations
• Test bulk update operations
• Test batch delete operations
• Validate batch size limits

Simple Implementation

from locust import task, HttpUser
import json
import random

class BatchOperationsUser(HttpUser):
    def on_start(self):
        self.created_items = []  # Track items we create

    @task(4)
    def test_batch_create_users(self):
        """Test creating multiple users in a single batch request"""
        
        # Create batch of 3-5 users
        batch_size = random.randint(3, 5)
        users_batch = []
        
        for i in range(batch_size):
            user = {
                "name": f"User {random.randint(1000, 9999)}",
                "email": f"user{random.randint(1000, 9999)}@example.com",
                "active": True
            }
            users_batch.append(user)
        
        batch_data = {"users": users_batch}
        
        with self.client.post(
            "/api/users/batch",
            json=batch_data,
            name="Batch Create Users"
        ) as response:
            if response.status_code in [200, 201]:
                try:
                    result = response.json()
                    created_count = len(result.get("created", []))
                    print(f"Batch create: {created_count}/{batch_size} users created")
                    
                    # Store created IDs for later use
                    if "created" in result:
                        for item in result["created"]:
                            if "id" in item:
                                self.created_items.append(item["id"])
                                
                except json.JSONDecodeError:
                    print("Batch create successful but no JSON response")
            else:
                response.failure(f"Batch create failed: {response.status_code}")

    @task(3)
    def test_batch_update_users(self):
        """Test updating multiple users in a single batch request"""
        
        if len(self.created_items) < 2:
            return  # Need at least 2 items to update
        
        # Select 2-3 items to update
        update_count = min(3, len(self.created_items))
        items_to_update = random.sample(self.created_items, update_count)
        
        updates_batch = []
        for item_id in items_to_update:
            update = {
                "id": item_id,
                "name": f"Updated User {random.randint(1000, 9999)}",
                "active": random.choice([True, False])
            }
            updates_batch.append(update)
        
        batch_data = {"updates": updates_batch}
        
        with self.client.put(
            "/api/users/batch",
            json=batch_data,
            name="Batch Update Users"
        ) as response:
            if response.status_code == 200:
                try:
                    result = response.json()
                    updated_count = len(result.get("updated", []))
                    print(f"Batch update: {updated_count}/{update_count} users updated")
                except json.JSONDecodeError:
                    print("Batch update successful")
            else:
                response.failure(f"Batch update failed: {response.status_code}")

    @task(2)
    def test_batch_create_products(self):
        """Test creating multiple products in batch"""
        
        batch_size = random.randint(2, 4)
        products_batch = []
        
        for i in range(batch_size):
            product = {
                "name": f"Product {random.randint(1000, 9999)}",
                "price": round(random.uniform(10.0, 100.0), 2),
                "category": random.choice(["electronics", "books", "clothing"])
            }
            products_batch.append(product)
        
        batch_data = {"products": products_batch}
        
        with self.client.post(
            "/api/products/batch",
            json=batch_data,
            name="Batch Create Products"
        ) as response:
            if response.status_code in [200, 201]:
                print(f"Batch created {batch_size} products")
            else:
                response.failure(f"Product batch create failed: {response.status_code}")

    @task(2)
    def test_large_batch_operation(self):
        """Test larger batch to check size limits"""
        
        # Try a larger batch (10 items)
        batch_size = 10
        large_batch = []
        
        for i in range(batch_size):
            item = {
                "name": f"Bulk Item {i}",
                "value": random.randint(1, 100),
                "type": "test"
            }
            large_batch.append(item)
        
        batch_data = {"items": large_batch}
        
        with self.client.post(
            "/api/items/batch",
            json=batch_data,
            name="Large Batch Operation"
        ) as response:
            if response.status_code in [200, 201]:
                print(f"Large batch ({batch_size} items) successful")
            elif response.status_code == 413:
                print(f"Large batch rejected - payload too large")
                response.failure("Batch size limit exceeded")
            elif response.status_code == 400:
                print(f"Large batch rejected - bad request")
                response.failure("Large batch validation failed")
            else:
                response.failure(f"Large batch failed: {response.status_code}")

    @task(1)
    def test_batch_delete(self):
        """Test deleting multiple items in batch"""
        
        if len(self.created_items) < 2:
            return  # Need items to delete
        
        # Delete 1-2 items
        delete_count = min(2, len(self.created_items))
        items_to_delete = []
        
        for _ in range(delete_count):
            if self.created_items:
                item_id = self.created_items.pop()
                items_to_delete.append(item_id)
        
        if not items_to_delete:
            return
        
        batch_data = {"ids": items_to_delete}
        
        with self.client.delete(
            "/api/users/batch",
            json=batch_data,
            name="Batch Delete Users"
        ) as response:
            if response.status_code in [200, 204]:
                print(f"Batch deleted {len(items_to_delete)} users")
            else:
                response.failure(f"Batch delete failed: {response.status_code}")

    @task(1)
    def test_empty_batch(self):
        """Test batch operation with empty data"""
        
        empty_batch = {"users": []}
        
        with self.client.post(
            "/api/users/batch",
            json=empty_batch,
            name="Empty Batch Test"
        ) as response:
            if response.status_code == 400:
                print("Empty batch correctly rejected")
            elif response.status_code in [200, 201]:
                print("Empty batch accepted (no-op)")
            else:
                response.failure(f"Empty batch unexpected response: {response.status_code}")

    @task(1)
    def test_mixed_batch_operation(self):
        """Test batch with mix of valid and invalid items"""
        
        mixed_batch = [
            {"name": "Valid User", "email": "valid@example.com"},
            {"name": "", "email": "invalid-email"},  # Invalid
            {"name": "Another Valid User", "email": "valid2@example.com"}
        ]
        
        batch_data = {"users": mixed_batch}
        
        with self.client.post(
            "/api/users/batch",
            json=batch_data,
            name="Mixed Batch Test"
        ) as response:
            if response.status_code in [200, 201]:
                try:
                    result = response.json()
                    created = len(result.get("created", []))
                    errors = len(result.get("errors", []))
                    print(f"Mixed batch: {created} created, {errors} errors")
                except json.JSONDecodeError:
                    print("Mixed batch processed")
            elif response.status_code == 400:
                print("Mixed batch rejected due to invalid items")
            else:
                response.failure(f"Mixed batch failed: {response.status_code}")

Setup Instructions

1. Replace API endpoints with your actual batch endpoints
2. Adjust batch sizes based on your API limits
3. Customize data structures for your specific entities
4. Update field names to match your API schema

What This Tests

• Batch Create: Creating multiple items in one request
• Batch Update: Updating multiple items in one request
• Batch Delete: Deleting multiple items in one request
• Size Limits: Testing batch size limitations
• Error Handling: Mixed valid/invalid items in batches

Batch Patterns

• All-or-Nothing: Entire batch fails if any item fails
• Partial Success: Some items succeed, others fail
• Best Effort: Process all valid items, report errors

Common Batch Endpoints

• POST /api/users/batch - Create multiple users
• PUT /api/users/batch - Update multiple users
• DELETE /api/users/batch - Delete multiple users
• POST /api/items/bulk - Bulk operations

Best Practices

• Keep batch sizes reasonable (10-100 items)
• Handle partial failures gracefully
• Provide clear error messages for failed items
• Consider batch processing timeouts