I was building a data processing system for a client that needed to continuously monitor stock prices from an API. The challenge was that network issues, API rate limits, and data format errors kept crashing my while loop.
I’ve learned that handling exceptions within while loops is crucial for building robust applications. Without proper exception handling, a single error can terminate your entire loop and crash your program.
In this article, I’ll share four proven methods I use to handle exceptions in while loops, complete with real-world examples that you can implement immediately.
Method 1 – Basic Try-Except Within While Loop
The simplest way to handle exceptions in a while loop is to wrap your code in a try-except block. This prevents the loop from terminating when an error occurs.
I use this method when I want the loop to continue running regardless of errors. Here’s how it works:
import time
import random
def process_customer_data():
"""Simulate processing customer data that might fail"""
customers = ["John Smith", "Sarah Johnson", "Mike Davis", "Lisa Wilson", "Tom Brown"]
counter = 0
while counter < 10:
try:
# Simulate random errors that might occur in real applications
if random.random() < 0.3: # 30% chance of error
raise ValueError("Database connection failed")
if random.random() < 0.2: # 20% chance of another error
raise ConnectionError("API timeout occurred")
# Process a random customer
customer = random.choice(customers)
processing_time = random.uniform(0.5, 2.0)
time.sleep(processing_time)
print(f"✓ Successfully processed customer: {customer}")
counter += 1
except ValueError as ve:
print(f"✗ Database error: {ve}")
except ConnectionError as ce:
print(f"✗ Network error: {ce}")
except Exception as e:
print(f"✗ Unexpected error: {e}")
# Small delay before next iteration
time.sleep(0.5)
print(f"\nCompleted processing {counter} customers")
# Run the function
process_customer_data()I executed the above example code and added the screenshot below.
This approach ensures that even if database connections fail or API calls timeout, your loop continues processing other customers.
Method 2 – Continue Statement with Exception Handling
Sometimes you want to skip the current iteration when an error occurs and immediately move to the next one. The continue statement is perfect for this scenario.
I frequently use this method when processing large datasets where some records might be corrupted or incomplete:
import json
import time
def process_sales_data():
"""Process sales data with error handling and continue statement"""
# Sample sales data that might have errors
sales_records = [
'{"customer": "Walmart", "amount": 15000, "date": "2024-01-15"}',
'{"customer": "Target", "amount": 12000, "date": "2024-01-16"}',
'{invalid_json}', # This will cause an error
'{"customer": "Amazon", "amount": 25000, "date": "2024-01-17"}',
'{"customer": "Costco", "amount": null}', # Missing amount
'{"customer": "Home Depot", "amount": 18000, "date": "2024-01-18"}',
]
total_sales = 0
processed_count = 0
current_index = 0
while current_index < len(sales_records):
try:
# Parse JSON data
record = json.loads(sales_records[current_index])
# Validate required fields
if not record.get('customer'):
raise ValueError("Missing customer name")
if not record.get('amount') or record['amount'] is None:
raise ValueError("Invalid sales amount")
# Process valid record
customer = record['customer']
amount = float(record['amount'])
total_sales += amount
processed_count += 1
print(f"✓ Processed: {customer} - ${amount:,.2f}")
except json.JSONDecodeError:
print(f"✗ Record {current_index + 1}: Invalid JSON format - skipping")
current_index += 1
continue
except ValueError as ve:
print(f"✗ Record {current_index + 1}: {ve} - skipping")
current_index += 1
continue
except Exception as e:
print(f"✗ Record {current_index + 1}: Unexpected error {e} - skipping")
current_index += 1
continue
current_index += 1
time.sleep(0.1) # Small delay for demonstration
print(f"\n--- Sales Summary ---")
print(f"Total records processed: {processed_count}")
print(f"Total sales: ${total_sales:,.2f}")
print(f"Average sale: ${total_sales/processed_count if processed_count > 0 else 0:,.2f}")
# Execute the function
process_sales_data()I executed the above example code and added the screenshot below.
The continue statement immediately jumps to the next iteration, skipping any remaining code in the current loop cycle.
This is particularly useful when you’re processing files or API responses where some data might be malformed.
Method 3 – Break Statement for Critical Errors
Not all errors should be ignored. Some exceptions indicate critical problems that require immediate attention and should stop the loop entirely.
I use the break statement when encountering errors that make it impossible or unsafe to continue:
import random
import time
from datetime import datetime
def monitor_server_health():
"""Monitor server health with critical error handling"""
max_attempts = 20
attempt_count = 0
consecutive_failures = 0
max_consecutive_failures = 3
print("Starting server health monitoring...")
print("=" * 50)
while attempt_count < max_attempts:
try:
attempt_count += 1
current_time = datetime.now().strftime("%H:%M:%S")
# Simulate server health check
cpu_usage = random.uniform(10, 95)
memory_usage = random.uniform(20, 90)
disk_usage = random.uniform(30, 85)
# Simulate different types of errors
if random.random() < 0.1: # 10% chance of network error
raise ConnectionError("Cannot connect to server")
if random.random() < 0.05: # 5% chance of authentication error
raise PermissionError("Authentication failed - invalid credentials")
if random.random() < 0.03: # 3% chance of critical system error
raise SystemError("Critical system failure detected")
# Check for concerning metrics
if cpu_usage > 90:
raise ResourceWarning(f"High CPU usage: {cpu_usage:.1f}%")
if memory_usage > 85:
raise ResourceWarning(f"High memory usage: {memory_usage:.1f}%")
# Successful health check
consecutive_failures = 0 # Reset failure counter
print(f"[{current_time}] ✓ Server OK - CPU: {cpu_usage:.1f}%, RAM: {memory_usage:.1f}%, Disk: {disk_usage:.1f}%")
except ConnectionError as ce:
consecutive_failures += 1
print(f"[{current_time}] ⚠ Network issue: {ce}")
if consecutive_failures >= max_consecutive_failures:
print(f"\n❌ CRITICAL: {consecutive_failures} consecutive network failures!")
print("Stopping monitoring - manual intervention required")
break
except PermissionError as pe:
print(f"[{current_time}] ❌ CRITICAL: {pe}")
print("Authentication failure - stopping monitoring")
break
except SystemError as se:
print(f"[{current_time}] 🚨 EMERGENCY: {se}")
print("Critical system error detected - immediate shutdown required")
break
except ResourceWarning as rw:
consecutive_failures += 1
print(f"[{current_time}] ⚠ Performance warning: {rw}")
except Exception as e:
consecutive_failures += 1
print(f"[{current_time}] ✗ Unexpected error: {e}")
time.sleep(2) # Wait 2 seconds between checks
print(f"\nMonitoring completed after {attempt_count} attempts")
if consecutive_failures >= max_consecutive_failures:
print("⚠ Action required: Check server status manually")
# Run server monitoring
monitor_server_health()I executed the above example code and added the screenshot below.
In this example, authentication failures and system errors trigger a break statement because continuing would be pointless or dangerous.
Network issues are treated as warnings, but too many consecutive failures also trigger a shutdown.
Method 4 – Advanced Exception Handling with Logging
For production applications, I always implement comprehensive logging alongside exception handling. This helps with debugging and monitoring application health.
Here’s my go-to approach for robust exception handling in while loops:
import logging
import time
import random
import json
from datetime import datetime
from typing import Dict, List
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('payment_processor.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class PaymentProcessor:
def __init__(self):
self.processed_payments = []
self.failed_payments = []
self.retry_queue = []
def process_payment_queue(self, payment_queue: List[Dict], max_runtime_minutes: int = 5):
"""Process payment queue with advanced exception handling"""
start_time = datetime.now()
max_runtime_seconds = max_runtime_minutes * 60
payment_index = 0
retry_attempts = {}
max_retries = 3
logger.info(f"Starting payment processing - {len(payment_queue)} payments in queue")
while payment_index < len(payment_queue):
# Check runtime limit
elapsed_time = (datetime.now() - start_time).total_seconds()
if elapsed_time > max_runtime_seconds:
logger.warning(f"Runtime limit reached ({max_runtime_minutes} minutes)")
break
current_payment = payment_queue[payment_index]
payment_id = current_payment.get('id', f'payment_{payment_index}')
try:
# Validate payment data
self._validate_payment(current_payment)
# Process payment (simulate API call)
result = self._process_single_payment(current_payment)
# Success
self.processed_payments.append({
'payment_id': payment_id,
'amount': current_payment['amount'],
'timestamp': datetime.now().isoformat(),
'result': result
})
logger.info(f"✓ Payment {payment_id} processed successfully - ${current_payment['amount']:.2f}")
except ValueError as ve:
# Data validation errors - don't retry
error_msg = f"Validation failed for payment {payment_id}: {ve}"
logger.error(error_msg)
self.failed_payments.append({
'payment_id': payment_id,
'error': str(ve),
'error_type': 'validation_error',
'timestamp': datetime.now().isoformat()
})
except ConnectionError as ce:
# Network errors - retry up to max_retries
retry_count = retry_attempts.get(payment_id, 0)
if retry_count < max_retries:
retry_attempts[payment_id] = retry_count + 1
logger.warning(f"Network error for payment {payment_id} (attempt {retry_count + 1}): {ce}")
# Add to retry queue and continue
self.retry_queue.append(current_payment)
time.sleep(2 ** retry_count) # Exponential backoff
continue
else:
logger.error(f"Payment {payment_id} failed after {max_retries} attempts: {ce}")
self.failed_payments.append({
'payment_id': payment_id,
'error': str(ce),
'error_type': 'network_error',
'retry_attempts': retry_count,
'timestamp': datetime.now().isoformat()
})
except PermissionError as pe:
# Authentication/authorization errors - critical
logger.critical(f"Authentication failed for payment {payment_id}: {pe}")
logger.critical("Stopping payment processing - check API credentials")
break
except Exception as e:
# Unexpected errors - log and continue
logger.error(f"Unexpected error processing payment {payment_id}: {e}", exc_info=True)
self.failed_payments.append({
'payment_id': payment_id,
'error': str(e),
'error_type': 'unexpected_error',
'timestamp': datetime.now().isoformat()
})
finally:
# This runs regardless of success or failure
payment_index += 1
# Progress update every 10 payments
if payment_index % 10 == 0:
logger.info(f"Progress: {payment_index}/{len(payment_queue)} payments processed")
# Generate summary report
self._generate_processing_report()
def _validate_payment(self, payment: Dict):
"""Validate payment data"""
required_fields = ['customer_id', 'amount', 'payment_method']
for field in required_fields:
if field not in payment or payment[field] is None:
raise ValueError(f"Missing required field: {field}")
if payment['amount'] <= 0:
raise ValueError("Payment amount must be positive")
if payment['amount'] > 10000:
raise ValueError("Payment amount exceeds daily limit")
def _process_single_payment(self, payment: Dict):
"""Simulate payment processing"""
# Simulate various error conditions
error_chance = random.random()
if error_chance < 0.1: # 10% network error
raise ConnectionError("Payment gateway timeout")
elif error_chance < 0.15: # 5% auth error
raise PermissionError("Invalid API key")
elif error_chance < 0.18: # 3% unexpected error
raise RuntimeError("Payment gateway internal error")
# Simulate processing time
time.sleep(random.uniform(0.1, 0.5))
return {
'transaction_id': f"txn_{random.randint(100000, 999999)}",
'status': 'completed',
'processing_time_ms': random.randint(100, 500)
}
def _generate_processing_report(self):
"""Generate comprehensive processing report"""
total_payments = len(self.processed_payments) + len(self.failed_payments)
success_rate = (len(self.processed_payments) / total_payments * 100) if total_payments > 0 else 0
total_amount = sum(p['amount'] for p in self.processed_payments)
logger.info("=" * 60)
logger.info("PAYMENT PROCESSING SUMMARY")
logger.info("=" * 60)
logger.info(f"Total Payments: {total_payments}")
logger.info(f"Successful: {len(self.processed_payments)}")
logger.info(f"Failed: {len(self.failed_payments)}")
logger.info(f"Success Rate: {success_rate:.1f}%")
logger.info(f"Total Amount Processed: ${total_amount:,.2f}")
logger.info(f"Payments in Retry Queue: {len(self.retry_queue)}")
# Error breakdown
if self.failed_payments:
error_types = {}
for failure in self.failed_payments:
error_type = failure['error_type']
error_types[error_type] = error_types.get(error_type, 0) + 1
logger.info("\nError Breakdown:")
for error_type, count in error_types.items():
logger.info(f" {error_type}: {count}")
# Example usage
def run_payment_processing_example():
"""Run the payment processing example"""
# Sample payment data
sample_payments = [
{'id': 'pay_001', 'customer_id': 'cust_123', 'amount': 99.99, 'payment_method': 'credit_card'},
{'id': 'pay_002', 'customer_id': 'cust_456', 'amount': 149.50, 'payment_method': 'debit_card'},
{'id': 'pay_003', 'customer_id': 'cust_789', 'amount': -50.00, 'payment_method': 'credit_card'}, # Invalid amount
{'id': 'pay_004', 'customer_id': 'cust_101', 'amount': 75.25, 'payment_method': 'paypal'},
{'id': 'pay_005', 'customer_id': None, 'amount': 200.00, 'payment_method': 'credit_card'}, # Missing customer_id
{'id': 'pay_006', 'customer_id': 'cust_202', 'amount': 125.75, 'payment_method': 'apple_pay'},
{'id': 'pay_007', 'customer_id': 'cust_303', 'amount': 300.00, 'payment_method': 'credit_card'},
{'id': 'pay_008', 'customer_id': 'cust_404', 'amount': 15000.00, 'payment_method': 'wire_transfer'}, # Exceeds limit
{'id': 'pay_009', 'customer_id': 'cust_505', 'amount': 89.99, 'payment_method': 'credit_card'},
{'id': 'pay_010', 'customer_id': 'cust_606', 'amount': 175.50, 'payment_method': 'debit_card'},
]
processor = PaymentProcessor()
processor.process_payment_queue(sample_payments, max_runtime_minutes=2)
# Run the example
run_payment_processing_example()This advanced approach combines multiple exception handling strategies with comprehensive logging and retry logic.
The finally block ensures that the loop index always increments, preventing infinite loops even when exceptions occur.
Conclusion
The four methods I’ve shared, basic try-except blocks, continue statements for recoverable errors, break statements for critical failures, and advanced logging with custom exceptions, have saved me countless hours of debugging and prevented numerous system crashes.
Remember these key principles: catch specific exceptions rather than using broad except clauses, use continue for recoverable errors that shouldn’t stop your loop, implement break for critical failures that require immediate attention, and always log your exceptions for easier troubleshooting.
You may also read:
- For loop vs while loop in Python
- Python For Loop with Index
- Use Python While with Assignment
- Python While Multiple Conditions
I am Bijay Kumar, a Microsoft MVP in SharePoint. Apart from SharePoint, I started working on Python, Machine learning, and artificial intelligence for the last 5 years. During this time I got expertise in various Python libraries also like Tkinter, Pandas, NumPy, Turtle, Django, Matplotlib, Tensorflow, Scipy, Scikit-Learn, etc… for various clients in the United States, Canada, the United Kingdom, Australia, New Zealand, etc. Check out my profile.
