Batch Processing
30-40% throughput improvement by batching database operations with graceful fallback.
When to Use This Skill
- Processing multiple related records (invoices, orders, events)
- Network latency is significant (cloud databases)
- Writes are independent (no inter-record dependencies)
- You can implement fallback for reliability
Core Concepts
Sequential processing is slow because each item requires multiple DB round trips. The solution is to collect all data first, then execute batch operations:
Sequential (slow):
Item 1 → DB → DB → DB
Item 2 → DB → DB → DB
Item 3 → DB → DB → DB
Batched (fast):
Item 1 → collect
Item 2 → collect
Item 3 → collect
All items → BATCH INSERTKey insight: Sequential mapping (fuzzy matching needs context), but batched writes (independent operations).
Implementation
Python
from decimal import Decimal
from typing import Dict, List
import time
class BatchProcessor:
"""
Batch-optimized processor with fallback
"""
def process_batch(self, items: List[Dict], user_id: str) -> Dict:
start_time = time.perf_counter()
# Collectors for batch operations
transactions_to_create = []
inventory_updates = {}
failed_items = []
items_processed = 0
# Step 1: Process mappings sequentially (context-dependent)
for idx, item in enumerate(items, 1):
try:
# Business logic that needs context
mapping = self.find_or_create_mapping(item)
# Collect for batch insert
transactions_to_create.append({
"user_id": user_id,
"item_id": mapping['item_id'],
"quantity": float(item['quantity']),
"unit_cost": float(item['unit_price']),
})
# Aggregate inventory updates by item
item_id = mapping['item_id']
if item_id not in inventory_updates:
inventory_updates[item_id] = Decimal('0')
inventory_updates[item_id] += Decimal(str(item['quantity']))
items_processed += 1
except Exception as e:
failed_items.append({
"line": idx,
"error": str(e)
})
continue
# Step 2: BATCH INSERT transactions
if transactions_to_create:
try:
self.client.table("transactions").insert(
transactions_to_create
).execute()
except Exception as e:
# CRITICAL: Fallback to sequential on batch failure
return self._fallback_to_sequential(items, user_id)
# Step 3: BATCH UPDATE inventory (aggregate first)
if inventory_updates:
self._batch_update_inventory(inventory_updates)
return {
"status": "partial_success" if failed_items else "success",
"items_processed": items_processed,
"items_failed": len(failed_items),
"failed_items": failed_items or None,
"processing_time_seconds": round(time.perf_counter() - start_time, 2)
}
def _batch_update_inventory(self, updates: Dict[str, Decimal]):
"""Batch query, individual updates (Supabase limitation)"""
item_ids = list(updates.keys())
# Get current quantities in one query
current = self.client.table("inventory").select(
"id, quantity"
).in_("id", item_ids).execute()
# Apply updates
for item in current.data:
item_id = item['id']
new_qty = Decimal(str(item['quantity'])) + updates[item_id]
self.client.table("inventory").update({
"quantity": float(new_qty)
}).eq("id", item_id).execute()
def _fallback_to_sequential(self, items: List[Dict], user_id: str) -> Dict:
"""Fallback ensures data integrity when batch fails"""
logger.warning("Falling back to sequential processing")
# Process one at a time
for item in items:
self.process_single(item, user_id)TypeScript
interface BatchResult {
status: 'success' | 'partial_success' | 'failed';
itemsProcessed: number;
itemsFailed: number;
failedItems?: { line: number; error: string }[];
processingTimeMs: number;
}
class BatchProcessor {
async processBatch(items: Item[], userId: string): Promise<BatchResult> {
const startTime = Date.now();
const transactionsToCreate: Transaction[] = [];
const inventoryUpdates = new Map<string, number>();
const failedItems: { line: number; error: string }[] = [];
let itemsProcessed = 0;
// Step 1: Process mappings sequentially
for (let idx = 0; idx < items.length; idx++) {
try {
const mapping = await this.findOrCreateMapping(items[idx]);
transactionsToCreate.push({
userId,
itemId: mapping.itemId,
quantity: items[idx].quantity,
unitCost: items[idx].unitPrice,
});
// Aggregate updates
const current = inventoryUpdates.get(mapping.itemId) || 0;
inventoryUpdates.set(mapping.itemId, current + items[idx].quantity);
itemsProcessed++;
} catch (error) {
failedItems.push({ line: idx + 1, error: error.message });
}
}
// Step 2: Batch insert
if (transactionsToCreate.length > 0) {
try {
await this.db.transactions.insertMany(transactionsToCreate);
} catch (error) {
return this.fallbackToSequential(items, userId);
}
}
// Step 3: Batch update inventory
await this.batchUpdateInventory(inventoryUpdates);
return {
status: failedItems.length > 0 ? 'partial_success' : 'success',
itemsProcessed,
itemsFailed: failedItems.length,
failedItems: failedItems.length > 0 ? failedItems : undefined,
processingTimeMs: Date.now() - startTime,
};
}
}Usage Examples
Invoice Processing
processor = BatchProcessor()
result = processor.process_batch(
items=invoice_data['line_items'],
user_id=user_id
)
if result['status'] == 'partial_success':
logger.warning(f"Some items failed: {result['failed_items']}")Best Practices
- Sequential mapping, batched writes - fuzzy matching needs context, writes don't
- Always implement fallback - batch operations can fail, sequential is reliable
- Aggregate before update - combine multiple updates to same record
- Handle partial success - one bad item shouldn't fail the entire batch
- Chunk large batches - 500 records max to avoid timeouts
Common Mistakes
- Batching operations that depend on each other's results
- No fallback when batch operations fail
- Not aggregating updates to the same record
- Collecting too many records before writing (memory pressure)
- Not logging individual items when batch fails (lose context)
Related Patterns
- checkpoint-resume - Resume processing after failures
- idempotency - Prevent duplicate processing on retry
- dead-letter-queue - Handle failed items