- Why error handling matters ā building reliable applications
- Common MySQL errors ā what can go wrong
- Try/except basics ā handling errors in Python
- Specific error handling ā dealing with different error types
- Retry logic ā handling temporary failures
- Logging errors ā tracking what went wrong
Why Error Handling Matters
When you're working with databases, things can go wrong. The database might be down, the connection might fail, or the data might be invalid. If you don't handle these errors, your program will crash and your users will have a bad experience.
š” Key concept: Error handling is about making your program resilient. Instead of crashing, it should handle errors gracefully and continue working.
Think of error handling like a safety net. When something goes wrong, the safety net catches you so you don't fall. In programming, error handling catches problems so your program doesn't crash.
What Happens Without Error Handling
# ============================================================
# WITHOUT ERROR HANDLING - Program Crashes!
# ============================================================
def get_student(student_id):
# No error handling
cursor.execute("SELECT * FROM students WHERE student_id = %s", (student_id,))
return cursor.fetchone()
# What happens if:
# - The database is down? š„ Program crashes!
# - student_id doesn't exist? š„ Program crashes!
# - There's a connection error? š„ Program crashes!
# ============================================================
# WITH ERROR HANDLING - Program Recovers!
# ============================================================
def get_student_safe(student_id):
try:
cursor.execute("SELECT * FROM students WHERE student_id = %s", (student_id,))
return cursor.fetchone()
except mysql.connector.Error as e:
print(f"Error getting student: {e}")
return None # Program continues!
# ============================================================
# THE DIFFERENCE
# ============================================================
print("""
Without error handling:
- Program crashes on any error
- Users see error messages
- Data might be corrupted
- Bad user experience
With error handling:
- Program handles errors gracefully
- Users see friendly messages
- Data stays consistent
- Better user experience
""")
Key point: Error handling makes your program robust and user-friendly. It prevents crashes and data corruption.
Quick Check: Why is error handling important? (Answer: It prevents crashes and makes applications more reliable)
Common MySQL Errors
What Can Go Wrong
Here are the most common errors you'll encounter when working with MySQL:
| Error Type | Error Code | Description | Common Cause |
|---|---|---|---|
| IntegrityError | 1062 | Duplicate entry | Inserting duplicate value in UNIQUE column |
| IntegrityError | 1452 | Foreign key constraint fails | Referencing a non-existent parent row |
| DataError | 1366 | Incorrect integer value | Putting text in a number column |
| OperationalError | 1049 | Unknown database | Database name doesn't exist |
| OperationalError | 2003 | Can't connect to MySQL server | MySQL isn't running or wrong host |
| OperationalError | 1045 | Access denied | Wrong username or password |
| ProgrammingError | 1146 | Table doesn't exist | Trying to use a table that doesn't exist |
| ProgrammingError | 1054 | Unknown column | Column name is misspelled or doesn't exist |
Error categories:
- IntegrityError ā data integrity violations (duplicates, foreign keys)
- DataError ā data type mismatches
- OperationalError ā connection and server issues
- ProgrammingError ā SQL syntax or structure errors
- InterfaceError ā database driver issues
Quick Check: What error occurs when you insert a duplicate value in a UNIQUE column? (Answer: IntegrityError with code 1062)
Try/Except Basics
The Foundation of Error Handling
In Python, you handle errors using try/except blocks. The code that might fail goes in the try block, and the error handling goes in the except block.
# ============================================================
# BASIC TRY/EXCEPT STRUCTURE
# ============================================================
try:
# Code that might cause an error
cursor.execute("SELECT * FROM students")
results = cursor.fetchall()
except mysql.connector.Error as e:
# Code that runs if there was an error
print(f"Database error: {e}")
# ============================================================
# EXAMPLE 1: Handling Connection Errors
# ============================================================
import mysql.connector
from mysql.connector import Error
try:
connection = mysql.connector.connect(
host="localhost",
user="root",
password="secret",
database="myapp_db"
)
print("Connected successfully")
except Error as e:
print(f"Connection failed: {e}")
finally:
if 'connection' in locals() and connection.is_connected():
connection.close()
print("Connection closed")
# ============================================================
# EXAMPLE 2: Handling Query Errors
# ============================================================
try:
cursor = connection.cursor()
cursor.execute("SELECT * FROM non_existent_table")
results = cursor.fetchall()
except mysql.connector.Error as e:
print(f"Query error: {e}")
print(f"Error code: {e.errno}")
print(f"SQL State: {e.sqlstate}")
# ============================================================
# EXAMPLE 3: Using else and finally
# ============================================================
try:
cursor.execute("SELECT * FROM students")
except Error as e:
print(f"Error: {e}")
else:
# Runs only if no exception occurred
print("Query executed successfully")
results = cursor.fetchall()
finally:
# Always runs, even if there was an error
print("Cleaning up...")
cursor.close()
connection.close()
Try/except structure:
- try ā code that might cause an error
- except ā code that runs if an error occurs
- else ā code that runs if no error occurs (optional)
- finally ā code that always runs (optional)
Quick Check: What is the purpose of the finally block? (Answer: It runs regardless of whether an error occurred or not, for cleanup)
Handling Specific Errors
Dealing with Different Error Types
Not all errors are the same. Handling different error types differently makes your code more robust and user-friendly.
# ============================================================
# HANDLING DIFFERENT ERROR TYPES
# ============================================================
from mysql.connector import Error, IntegrityError, DataError, OperationalError, ProgrammingError
def insert_student(first_name, last_name, age, email):
try:
cursor.execute(
"INSERT INTO students (first_name, last_name, age, email) VALUES (%s, %s, %s, %s)",
(first_name, last_name, age, email)
)
connection.commit()
return {"success": True, "message": "Student added"}
except IntegrityError as e:
# Duplicate entries, foreign key violations
if e.errno == 1062:
return {"success": False, "message": "Email already exists"}
elif e.errno == 1452:
return {"success": False, "message": "Referenced record not found"}
else:
return {"success": False, "message": f"Integrity error: {e}"}
except DataError as e:
# Data type mismatches
return {"success": False, "message": f"Invalid data type: {e}"}
except ProgrammingError as e:
# SQL syntax errors, missing tables
return {"success": False, "message": f"SQL error: {e}"}
except OperationalError as e:
# Connection issues, server errors
return {"success": False, "message": f"Database operational error: {e}"}
except Error as e:
# Any other MySQL error
return {"success": False, "message": f"Database error: {e}"}
except Exception as e:
# Any other Python error
return {"success": False, "message": f"Unexpected error: {e}"}
# ============================================================
# HANDLING SPECIFIC ERROR CODES
# ============================================================
from mysql.connector import errorcode
try:
cursor.execute("INSERT INTO students (email) VALUES (%s)", ("existing@email.com",))
connection.commit()
except mysql.connector.Error as e:
if e.errno == errorcode.ER_DUP_ENTRY:
print("Duplicate email! Please use a different email.")
elif e.errno == errorcode.ER_BAD_DB_ERROR:
print("Database does not exist.")
elif e.errno == errorcode.ER_ACCESS_DENIED_ERROR:
print("Access denied. Check your username and password.")
else:
print(f"Other error: {e}")
Specific handling benefits:
- User-friendly messages ā show meaningful error messages
- Appropriate responses ā different errors need different handling
- Better debugging ā know exactly what went wrong
- Data integrity ā prevent corrupted data
Quick Check: Why should you handle different error types differently? (Answer: Different errors require different responses and user messages)
Retry Logic
Handling Temporary Failures
Some errors are temporary ā the database might be busy, or there might be a network glitch. In these cases, it makes sense to retry the operation.
# ============================================================
# BASIC RETRY LOGIC
# ============================================================
import time
def retry_operation(operation, max_retries=3, delay=2):
"""Retry an operation multiple times"""
for attempt in range(max_retries):
try:
return operation()
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
print(f"Retrying in {delay} seconds...")
time.sleep(delay)
else:
print("All retries failed")
raise
# ============================================================
# RETRY FOR DATABASE OPERATIONS
# ============================================================
def connect_with_retry(db_config, max_retries=5, delay=2):
"""Connect to MySQL with retry logic"""
def connect():
return mysql.connector.connect(**db_config)
for attempt in range(max_retries):
try:
connection = connect()
print("Connected successfully")
return connection
except mysql.connector.Error as e:
print(f"Connection attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
print(f"Retrying in {delay} seconds...")
time.sleep(delay)
else:
print("All connection attempts failed")
raise
# ============================================================
# RETRY FOR QUERY EXECUTION
# ============================================================
def execute_with_retry(query, params=None, max_retries=3, delay=1):
"""Execute a query with retry logic"""
for attempt in range(max_retries):
try:
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
connection.commit()
return cursor.rowcount
except mysql.connector.OperationalError as e:
# Network issues, deadlocks, etc.
print(f"Query attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
print(f"Retrying in {delay} seconds...")
time.sleep(delay)
else:
raise
except Exception as e:
# Don't retry non-transient errors
print(f"Non-retryable error: {e}")
raise
# ============================================================
# SMART RETRY WITH EXPONENTIAL BACKOFF
# ============================================================
def smart_retry(operation, max_retries=5, base_delay=1):
"""Retry with exponential backoff"""
for attempt in range(max_retries):
try:
return operation()
except Exception as e:
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt) # 1, 2, 4, 8, 16 seconds
print(f"Attempt {attempt + 1} failed. Retrying in {delay} seconds...")
time.sleep(delay)
Retry logic benefits:
- Fault tolerance ā handles temporary failures
- Better user experience ā operations succeed eventually
- Exponential backoff ā prevents overwhelming the database
- Transient errors ā network issues, server busy
Quick Check: When should you use retry logic? (Answer: For transient errors like network issues or temporary server problems)
Logging Errors
Tracking What Goes Wrong
Logging errors is essential for debugging and monitoring. It helps you understand what's happening in your application and fix problems before they affect users.
# ============================================================
# SETTING UP LOGGING
# ============================================================
import logging
import datetime
# Basic logging configuration
logging.basicConfig(
level=logging.ERROR,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('database_errors.log'),
logging.StreamHandler() # Also print to console
]
)
logger = logging.getLogger('database')
# ============================================================
# LOGGING DATABASE ERRORS
# ============================================================
def log_database_error(error, operation, params=None):
"""Log database errors with context"""
log_data = {
'timestamp': datetime.datetime.now().isoformat(),
'operation': operation,
'params': params,
'error_code': error.errno if hasattr(error, 'errno') else None,
'error_message': str(error)
}
# Log as JSON or structured format
logger.error(f"Database error: {log_data}")
# ============================================================
# USING LOGGING IN YOUR CODE
# ============================================================
from mysql.connector import Error
def insert_with_logging(first_name, last_name, email):
try:
query = "INSERT INTO students (first_name, last_name, email) VALUES (%s, %s, %s)"
cursor.execute(query, (first_name, last_name, email))
connection.commit()
logger.info(f"Inserted student: {first_name} {last_name}")
return True
except Error as e:
# Log the error with context
log_database_error(e, "INSERT", {
'first_name': first_name,
'last_name': last_name,
'email': email
})
# Show user-friendly message
if e.errno == 1062:
print("Email already exists")
else:
print("An error occurred while adding the student")
return False
# ============================================================
# LOGGING ALL OPERATIONS
# ============================================================
class LoggedDatabase:
def __init__(self, connection):
self.connection = connection
self.cursor = connection.cursor()
self.logger = logging.getLogger(__name__)
def execute(self, query, params=None):
"""Execute a query with logging"""
try:
if params:
self.cursor.execute(query, params)
else:
self.cursor.execute(query)
self.logger.info(f"Query executed: {query[:100]}...")
return self.cursor
except Error as e:
self.logger.error(f"Query failed: {query}")
self.logger.error(f"Error: {e}")
raise
Logging benefits:
- Debugging ā see what went wrong and when
- Monitoring ā track error patterns
- Auditing ā keep records of database operations
- Fix problems faster ā logs tell you exactly what happened
Quick Check: Why is logging errors important? (Answer: It helps with debugging, monitoring, and fixing problems)
Real-World Example: Robust Database Service
Building a Resilient Database Service
# ============================================================
# COMPLETE ROBUST DATABASE SERVICE
# ============================================================
import mysql.connector
from mysql.connector import Error, IntegrityError, DataError, OperationalError, ProgrammingError
import logging
import time
import datetime
class RobustDatabaseService:
"""A database service with comprehensive error handling"""
def __init__(self, db_config):
self.db_config = db_config
self.connection = None
self.cursor = None
self.logger = self._setup_logger()
self.max_retries = 3
self.retry_delay = 1
def _setup_logger(self):
"""Set up logging configuration"""
logger = logging.getLogger('DatabaseService')
logger.setLevel(logging.INFO)
# File handler
file_handler = logging.FileHandler('db_service.log')
file_handler.setFormatter(
logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
)
logger.addHandler(file_handler)
return logger
def connect(self):
"""Connect with retry logic"""
for attempt in range(self.max_retries):
try:
self.connection = mysql.connector.connect(**self.db_config)
self.cursor = self.connection.cursor()
self.logger.info("Connected to database")
return True
except Error as e:
self.logger.warning(f"Connection attempt {attempt+1} failed: {e}")
if attempt < self.max_retries - 1:
time.sleep(self.retry_delay * (attempt + 1))
else:
self.logger.error("All connection attempts failed")
return False
def execute_query(self, query, params=None, retry=True):
"""Execute a query with error handling and optional retry"""
for attempt in range(self.max_retries if retry else 1):
try:
if not self.connection or not self.connection.is_connected():
self.connect()
if params:
self.cursor.execute(query, params)
else:
self.cursor.execute(query)
self.connection.commit()
return self.cursor
except IntegrityError as e:
# Data integrity errors - usually don't retry
self.logger.error(f"Integrity error: {e}")
if e.errno == 1062:
raise ValueError("Duplicate entry")
elif e.errno == 1452:
raise ValueError("Referenced record not found")
else:
raise
except DataError as e:
# Data type errors - don't retry
self.logger.error(f"Data error: {e}")
raise ValueError(f"Invalid data: {e}")
except ProgrammingError as e:
# SQL syntax errors - don't retry
self.logger.error(f"SQL error: {e}")
raise SyntaxError(f"SQL error: {e}")
except OperationalError as e:
# Operational errors - can retry
self.logger.warning(f"Operational error (attempt {attempt+1}): {e}")
if attempt < self.max_retries - 1:
time.sleep(self.retry_delay * (attempt + 1))
self.connect()
else:
raise
except Error as e:
# Other database errors
self.logger.error(f"Database error: {e}")
raise
except Exception as e:
# Unexpected errors
self.logger.error(f"Unexpected error: {e}")
raise
def insert_student(self, first_name, last_name, age, email):
"""Insert a student with comprehensive error handling"""
try:
query = """
INSERT INTO students (first_name, last_name, age, email)
VALUES (%s, %s, %s, %s)
"""
params = (first_name, last_name, age, email)
cursor = self.execute_query(query, params)
rowcount = cursor.rowcount
self.logger.info(f"Inserted student: {first_name} {last_name}")
return {
"success": True,
"message": "Student added successfully",
"rowcount": rowcount
}
except ValueError as e:
# Handle validation errors
return {"success": False, "message": str(e)}
except Exception as e:
# Handle other errors
self.logger.error(f"Failed to insert student: {e}")
return {"success": False, "message": f"Database error: {e}"}
def get_student(self, student_id):
"""Get a student by ID with error handling"""
try:
query = "SELECT * FROM students WHERE student_id = %s"
cursor = self.execute_query(query, (student_id,))
result = cursor.fetchone()
if result:
return {"success": True, "data": result}
else:
return {"success": False, "message": "Student not found"}
except Exception as e:
self.logger.error(f"Failed to get student: {e}")
return {"success": False, "message": f"Database error: {e}"}
def update_student(self, student_id, updates):
"""Update a student with error handling"""
try:
set_clause = []
params = []
for key, value in updates.items():
set_clause.append(f"{key} = %s")
params.append(value)
params.append(student_id)
query = f"UPDATE students SET {', '.join(set_clause)} WHERE student_id = %s"
cursor = self.execute_query(query, params)
if cursor.rowcount > 0:
self.logger.info(f"Updated student {student_id}")
return {"success": True, "message": "Student updated"}
else:
return {"success": False, "message": "Student not found"}
except Exception as e:
self.logger.error(f"Failed to update student: {e}")
return {"success": False, "message": f"Database error: {e}"}
def delete_student(self, student_id):
"""Delete a student with error handling"""
try:
query = "DELETE FROM students WHERE student_id = %s"
cursor = self.execute_query(query, (student_id,))
if cursor.rowcount > 0:
self.logger.info(f"Deleted student {student_id}")
return {"success": True, "message": "Student deleted"}
else:
return {"success": False, "message": "Student not found"}
except Exception as e:
self.logger.error(f"Failed to delete student: {e}")
return {"success": False, "message": f"Database error: {e}"}
def close(self):
"""Clean up resources"""
if self.cursor:
self.cursor.close()
if self.connection and self.connection.is_connected():
self.connection.close()
self.logger.info("Database connection closed")
# ============================================================
# DEMONSTRATION
# ============================================================
db_config = {
"host": "localhost",
"user": "root",
"password": "secret",
"database": "myapp_db"
}
service = RobustDatabaseService(db_config)
if service.connect():
# 1. Insert a student
print("\nInserting student...")
result = service.insert_student("Rahul", "Sharma", 22, "rahul@email.com")
print(f"Result: {result}")
# 2. Insert duplicate
print("\nInserting duplicate email...")
result = service.insert_student("Duplicate", "User", 25, "rahul@email.com")
print(f"Result: {result}")
# 3. Get a student
print("\nGetting student...")
result = service.get_student(1)
print(f"Result: {result}")
# 4. Update a student
print("\nUpdating student...")
result = service.update_student(1, {"age": 23, "email": "rahul_new@email.com"})
print(f"Result: {result}")
service.close()
This robust service demonstrates:
- Comprehensive error handling for all operations
- Retry logic with exponential backoff
- Logging for debugging and monitoring
- User-friendly error messages
- Transaction management
- Resource cleanup
Quick Check: What makes a database service "robust"? (Answer: Comprehensive error handling, retry logic, logging, and graceful failure recovery)
Best Practices
Error Handling Guidelines
# ============================================================
# BEST PRACTICES FOR ERROR HANDLING
# ============================================================
print("1. ALWAYS USE TRY/EXCEPT")
print(" - Wrap database operations in try/except")
print(" - Never let errors crash your program")
print("\n2. HANDLE SPECIFIC ERRORS")
print(" - Handle different error types differently")
print(" - Use specific error classes")
print("\n3. PROVIDE USER-FRIENDLY MESSAGES")
print(" - Don't show raw error messages to users")
print(" - Provide meaningful, actionable messages")
print("\n4. LOG ERRORS FOR DEBUGGING")
print(" - Log all errors with context")
print(" - Include timestamps and operation details")
print("\n5. USE RETRY LOGIC")
print(" - Retry transient errors")
print(" - Use exponential backoff")
print("\n6. CLEAN UP RESOURCES")
print(" - Always close connections")
print(" - Use try/finally for cleanup")
print("\n7. VALIDATE DATA EARLY")
print(" - Validate data before database operations")
print(" - Catch errors before they reach the database")
print("\n8. USE TRANSACTIONS")
print(" - Group related operations in transactions")
print(" - Rollback on errors")
print("\n9. TEST ERROR HANDLING")
print(" - Test with different error scenarios")
print(" - Make sure error handling works")
print("\n10. MONITOR ERRORS")
print(" - Track error rates")
print(" - Set up alerts for critical errors")
Summary of best practices:
- Always use try/except ā never let errors crash your program
- Handle specific errors ā different errors need different handling
- Provide user-friendly messages ā don't show raw errors
- Log errors ā for debugging and monitoring
- Use retry logic ā for transient errors
- Clean up resources ā always close connections
- Validate data early ā catch errors before the database
Quick Check: What is the most important rule for error handling? (Answer: Always use try/except to handle errors gracefully)
Try It Yourself
Experiment with error handling in the editor below.
ERROR HANDLING - PRACTICE
========================================
1. SUCCESSFUL INSERT
----------------------------------------
Before insert:
-------------------------------------------------------
ID: 1 | Rahul Sharma | Age: 22 | rahul@email.com
ID: 2 | Priya Patel | Age: 25 | priya@email.com
Total: 2 students
Inserted: Amit Singh
Result: {'success': True, 'id': 3}
After insert:
-------------------------------------------------------
ID: 1 | Rahul Sharma | Age: 22 | rahul@email.com
ID: 2 | Priya Patel | Age: 25 | priya@email.com
ID: 3 | Amit Singh | Age: 24 | amit@email.com
Total: 3 students
2. DUPLICATE EMAIL ERROR
----------------------------------------
Error: Duplicate email
Result: {'success': False, 'message': 'Duplicate email'}
3. INVALID AGE ERROR
----------------------------------------
Error: Age cannot exceed 150
Result: {'success': False, 'message': 'Age cannot exceed 150'}
4. NON-INTEGER AGE ERROR
----------------------------------------
Error: Age must be a positive integer
Result: {'success': False, 'message': 'Age must be a positive integer'}
Error handling makes your application robust!
You've Got It!
You now know how to handle MySQL errors in Python. You understand try/except blocks, specific error types, retry logic, and logging.
Quick Quiz
Test what you've learned:
Frequently Asked Questions
What is the difference between IntegrityError and DataError?
Should I always retry on any error?
What is exponential backoff?
What is a common interview question about error handling?
How can I test my error handling?
Where to Go From Here
Now that you know how to handle database errors, check out these related topics:
Connection Pooling
Learn how to manage database connections efficiently.
Learn More āParameterized Queries
Learn how to keep your database secure.
Learn More āTransactions
Learn how to group operations safely.
Learn More ā