Python

P
PythonHub
Progress 0%

Python

  • Home
  • History of Python
  • Applications of Python
  • Introduction To Python
    • What is Python
    • Character Set
    • Tokens in Python
    • Python Execution Mode
    • Variable And Identifiers
    • Data Types in Python
    • Operators And Expressions
    • Constants in Python
    • Assignment Statement
    • Input / Output in Python
    • Simple Python Scripts
    • Namespace in Python
    • šŸ“ Assignments
  • Operators in Python
    • Arithmetic Operators
    • Assignment Operators
    • Shorthand Assignment Operators
    • Relational Operators
    • Logical Operators
    • Bitwise Operators
    • Special Operators
    • šŸ“ Assignments
  • Input Output in Python
    • Accept Input
    • Output Formatting
    • šŸ“ Assignments
  • Conditional Statement
    • Decision Making
    • if Statement
    • IF-ELSE STATEMENT
    • IF-ELSE LADDER
    • NESTED IF-ELSE
    • Short Hand IF-ELSE
    • šŸ“ Assignments
  • Loops
    • Introduction to Loops
    • While Loop
    • Nested While Loop
    • šŸ“ While Loop Assignments
    • For Loop
    • For Loop Examples
    • Nested For Loop
    • Nested For Loop Examples
    • Infinite While Loops
    • Infinite For Loops
    • Break, Continue and Else in Loops
    • Difference Between For and While Loop
    • For Each Loop
    • šŸ“ For Each Assignments
    • šŸ“ All Loops Assignments
  • List
    • List in Python
    • Access List Elements
    • List Functions
    • Iterate (Loop) List
    • List Comprehension
    • šŸ“ Assignments
  • Tuple
    • Tuple in Python
    • Access Tuple Elements
    • Tuple Functions
    • Iterate (Loop) Tuple
    • Unpack Tuple
    • Tuple Comprehension
    • šŸ“ Assignments
  • Set
    • Set in Python
    • Access Set Elements
    • Set Methods
    • Iterate (Loop) Set
    • Pack/Unpack Set
    • Set Comprehension
    • šŸ“ Assignments
  • Dictionary
    • Dictionary
    • Access Dictionary Items
    • Dictionary Methods
    • Iterate (Loop) Dictionary
    • Formatting Dictionaries
    • Nested Dictionaries
    • Dictionary Comprehension
    • šŸ“ Assignments
  • Diff List Tuple Set Dictionary
    • List vs Tuple
    • List vs Set
    • List vs Dictionary
    • Tuple vs Set
    • Tuple vs Dictionary
    • Dictionary vs Set
    • šŸ“ Assignments
  • Exception
    • Error vs Exception
    • Types of Exception
    • Exception Handling
    • User Defined Exception
    • Logging Exception
    • šŸ“ Assignments
  • Functions
    • Introduction to Functions
    • Modular Programming
    • Types of Functions
    • Inbuilt Functions
    • Need For User-Defined Function
    • Elements of User Defined Function
    • Function Arguments
    • Nesting of Functions
    • Recursion
    • Global Local and Non Local
    • Python Lambda Functions
    • šŸ“ Assignments
  • Python Module
    • Introduction to Module
    • Inbuilt Modules in Python
    • User Defined Module
    • šŸ“ Assignments
  • File Handling
    • Introduction to Files
    • Create File
    • Read Files
    • Write to File
    • Rename File
    • Copy File
    • Move File
    • List Files in Directory
    • Binary Files
    • Zipping and Unzipping Files
    • šŸ“ Assignments
  • Strings
    • Basics of Strings
    • String Special Operators
    • String Formatting Operators
    • String Methods
    • šŸ“ Assignments
  • Regular Expressions
  • Python OOPS
    • Basics of Object Oriented
    • What are Classes and Objects?
    • Creating Class and Object
    • OOP vs Procedural Programming
    • Difference Between Classes and Objects
    • Constructors
    • Destructor
    • Built Class Methods and Attributes
    • Class and Instance Variables
    • Inheritance in Python
    • Single Inheritance
    • Multiple Inheritance
    • Multilevel Inheritance
    • Hierarchical Inheritance
    • Hybrid Inheritance
    • Abstraction
    • Method Overriding
    • Abstract Method
    • Interfaces in Python
    • Abstract Class vs Interface
    • Public, Private and Protected
    • Overloading vs Overriding
    • Inheritance vs Composition
    • Encapsulation
    • Polymorphism
    • Inner Classes
    • šŸ“ Assignments
  • Advanced Python
    • Iterator in Python
    • Generator in Python
    • Decorator in Python
    • Type Hints in Python
    • Async/Await Programming
    • Dataclasses in Python
    • Context Managers in Python
    • Property Decorator in Python
    • __slots__ in Python
    • Enums in Python
    • Walrus Operator in Python
    • Match-Case in Python
    • šŸ“ Assignments
  • Python Standard Library
    • Collections Module
    • Itertools Module
    • Functools Module
    • Datetime Module
    • JSON Module
    • OS Module
    • Sys Module
    • Random Module
    • Math Module
    • šŸ“ Assignments
  • Python Testing
    • Unit Testing in Python
    • Pytest Framework
    • Mocking in Python
    • šŸ“ Assignments
  • Python Best Practices
    • PEP 8 Style Guide
    • Docstrings in Python
    • Logging in Python
    • Code Optimization Tips
    • Debugging Techniques
    • šŸ“ Assignments
  • MySQL Database in Python
    • Introduction to MySQL with Python
    • DBMS vs File System
    • Connecting to MySQL Database
    • Create Database in MySQL
    • Create Table in MySQL
    • Insert Data in MySQL
    • Insert Multiple Rows
    • Select Data from MySQL
    • WHERE Clause in MySQL
    • Update Data in MySQL
    • Delete Data from MySQL
    • Parameterized Queries
    • Transaction Management
    • Error Handling
    • Connection Pooling
    • MySQL Drivers Guide
    • Joins in MySQL
    • Aggregation Functions
    • Backup MySQL Database
    • Best Practices
    • šŸ“ Assignments
  • MySQL Database Operations
    • SELECT Statement
    • MySQL Operators
    • DDL Statements
    • DML Statements
    • Subqueries
    • JOIN Operations
    • Aggregation
    • Case Study
    • šŸ“ Assignments
  • Graphics in Python
  • Threads in Python
    • Introduction to Threads
    • Process vs Threads
    • Concurrent Programming & GIL
    • Uses of Threads
    • Creating Threads
    • Single Tasking
    • Multi Tasking
    • Thread Synchronization
    • šŸ“ Assignments
  • Interview Questions & Answers
  • Python Case Studies
  • Multiple Choice Questions
  • šŸ“ Practice Problems
Get in Touch
  • tech2dsm@gmail.com

© Sankalan Data Tech

Python Language Interactive Tutorial

Python: Error Handling in MySQL

Python MySQL Error Handling - Complete Guide

Learn how to handle MySQL errors gracefully and build robust applications.

Created by Sankalan Data Tech Team Verified
Data Engineers, Analysts, Scientists & Trainers
Created by experienced Python developers, data engineers, and data scientists to make programming easy through practical examples, real-world experience, and clear explanations.
On this page:
  • Why Error Handling Matters
  • Common MySQL Errors
  • Try/Except Basics
  • Handling Specific Errors
  • Retry Logic
  • Logging Errors
  • Real-World Example
  • Best Practices
  • Try It Yourself
  • Quiz
  • FAQ
Share this tutorial:
Twitter LinkedIn Facebook WhatsApp Reddit Telegram Email Copy Link
What You'll Learn Here
  • 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.

1

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

2

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

3

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

4

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

5

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

6

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

7

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

8

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.

Loading Pyodide... 0%
Python Code Editor
========================================
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:

1. What is the purpose of error handling in database applications?
2. What error occurs when you insert a duplicate value in a UNIQUE column?
3. What is the purpose of the finally block in try/except?
4. When should you use retry logic?
5. Why is logging errors important?

Frequently Asked Questions

What is the difference between IntegrityError and DataError? ā–¼

IntegrityError occurs when data violates database constraints (duplicate entries, foreign key violations). DataError occurs when there's a data type mismatch (putting text in a number column).

Should I always retry on any error? ā–¼

No, only retry on transient errors like network issues or temporary server problems. Don't retry on data errors or syntax errors because retrying won't fix them.

What is exponential backoff? ā–¼

Exponential backoff is a retry strategy where the delay between retries increases exponentially (1s, 2s, 4s, 8s, 16s). This prevents overwhelming the server when it's struggling.

What is a common interview question about error handling? ā–¼

Common questions include: "How do you handle database errors in Python?" "What is the difference between try/except and try/except/else/finally?" and "How would you implement retry logic for database operations?"

How can I test my error handling? ā–¼

You can test error handling by simulating different error scenarios: turn off the database, use invalid data, try to duplicate data, and use incorrect SQL syntax. This helps ensure your error handling works properly.

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 →
Interview Resources
  • Python Syntax & Variables Interview Questions
  • Top SQL Interview Questions & Answers
  • SQL Joins: Displaying Data from Multiple Tables FAQ
  • Python Lists and Dictionaries Interview Questions
  • Python OOP Interview Questions
  • SQL Set Operators Interview Questions
Previous: Transaction Management Next: Connection Pooling →