Error Handling

class MySQLError(Exception):
    """
    Base exception for all MySQL-related errors.
    
    Attributes:
    - errno (int): MySQL error number (-1 if not applicable)
    - errmsg (str): Error message description
    """
    
    def __init__(self, *args):
        """
        Initialize MySQL error.
        
        Parameters:
        - args: Error number and message, or just message
        """

class Warning(Warning, MySQLError):
    """
    Exception raised for important warnings like data truncations
    while inserting, etc.
    
    Used for non-fatal issues that don't prevent operation completion
    but indicate potential problems.
    """

class Error(MySQLError):
    """
    Exception that is the base class of all other error exceptions
    (not Warning).
    
    Base class for all serious database errors.
    """

class InterfaceError(Error):
    """
    Exception raised for errors that are related to the database
    interface rather than the database itself.
    
    Examples:
    - Connection parameter errors
    - Protocol violations
    - Client library issues
    """

class DatabaseError(Error):
    """
    Exception raised for errors that are related to the database.
    
    Base class for all database-related errors.
    """

class DataError(DatabaseError):
    """
    Exception raised for errors that are due to problems with the
    processed data like division by zero, numeric value out of range, etc.
    
    Examples:
    - Data truncation warnings
    - Invalid data format
    - Out of range values
    """

class OperationalError(DatabaseError):
    """
    Exception raised for errors that are related to the database's
    operation and not necessarily under the control of the programmer,
    e.g. an unexpected disconnect occurs, the data source name is not
    found, a transaction could not be processed, a memory allocation
    error occurred during processing, etc.
    
    Examples:
    - Connection lost
    - Database access denied
    - Lock deadlock
    - Server shutdown
    """

class IntegrityError(DatabaseError):
    """
    Exception raised when the relational integrity of the database
    is affected, e.g. a foreign key check fails, duplicate key, etc.
    
    Examples:
    - Duplicate key violations
    - Foreign key constraint failures
    - NOT NULL constraint violations
    """

class InternalError(DatabaseError):
    """
    Exception raised when the database encounters an internal
    error, e.g. the cursor is not valid anymore, the transaction is
    out of sync, etc.
    
    Examples:
    - Cursor state errors
    - Transaction state errors
    - Internal server errors
    """

class ProgrammingError(DatabaseError):
    """
    Exception raised for programming errors, e.g. table not found
    or already exists, syntax error in the SQL statement, wrong number
    of parameters specified, etc.
    
    Examples:
    - SQL syntax errors
    - Table/column doesn't exist
    - Wrong number of parameters
    - Database doesn't exist
    """

class NotSupportedError(DatabaseError):
    """
    Exception raised in case a method or database API was used
    which is not supported by the database, e.g. requesting a
    .rollback() on a connection that does not support transaction or
    has transactions turned off.
    
    Examples:
    - Unsupported SQL features
    - Disabled functionality
    - Version compatibility issues
    """

def raise_mysql_exception(data):
    """
    Parse MySQL error packet and raise appropriate exception.
    
    Parameters:
    - data (bytes): Raw error packet data from MySQL server
    
    Raises:
    Appropriate MySQL exception based on error code
    """

def _get_error_info(data):
    """
    Extract error information from MySQL error packet.
    
    Parameters:
    - data (bytes): Raw error packet data
    
    Returns:
    tuple: (errno, sqlstate, errorvalue)
    """

def _check_mysql_exception(errinfo):
    """
    Map MySQL error code to appropriate exception class and raise it.
    
    Parameters:
    - errinfo (tuple): Error information (errno, sqlstate, errorvalue)
    
    Raises:
    Appropriate MySQL exception class
    """

import cymysql
from cymysql import (
    OperationalError, ProgrammingError, IntegrityError,
    DataError, InternalError, InterfaceError
)

try:
    conn = cymysql.connect(
        host='localhost',
        user='invalid_user',
        password='wrong_password',
        db='nonexistent_db'
    )
except OperationalError as e:
    print(f"Connection failed: {e}")
    print(f"Error code: {e.errno}")
    print(f"Error message: {e.errmsg}")
except InterfaceError as e:
    print(f"Interface error: {e}")

import cymysql
from cymysql import ProgrammingError, IntegrityError, DataError

conn = cymysql.connect(host='localhost', user='root', password='', db='test')
cursor = conn.cursor()

try:
    # SQL syntax error
    cursor.execute("SELCT * FROM users")  # Typo in SELECT
except ProgrammingError as e:
    print(f"SQL syntax error: {e}")

try:
    # Table doesn't exist
    cursor.execute("SELECT * FROM nonexistent_table")
except ProgrammingError as e:
    print(f"Table not found: {e}")

try:
    # Duplicate key error
    cursor.execute("INSERT INTO users (id, email) VALUES (1, 'test@example.com')")
    cursor.execute("INSERT INTO users (id, email) VALUES (1, 'duplicate@example.com')")
except IntegrityError as e:
    print(f"Integrity constraint violation: {e}")

try:
    # Data too long
    cursor.execute("INSERT INTO users (name) VALUES (%s)", ('x' * 1000,))
except DataError as e:
    print(f"Data error: {e}")

cursor.close()
conn.close()

import cymysql
from cymysql import OperationalError, InternalError

conn = cymysql.connect(host='localhost', user='root', password='', db='test')

try:
    # Simulate connection loss
    cursor = conn.cursor()
    
    # Check if connection is alive
    conn.ping()
    
    cursor.execute("SELECT SLEEP(1)")
    result = cursor.fetchone()
    
except OperationalError as e:
    print(f"Connection lost: {e}")
    # Try to reconnect
    try:
        conn.ping(reconnect=True)
        print("Reconnection successful")
    except OperationalError:
        print("Reconnection failed")

except InternalError as e:
    print(f"Internal database error: {e}")

finally:
    try:
        cursor.close()
        conn.close()
    except:
        pass

import cymysql
from cymysql import OperationalError, IntegrityError, InternalError

conn = cymysql.connect(host='localhost', user='root', password='', db='test')
cursor = conn.cursor()

try:
    # Start transaction
    conn.autocommit(False)
    
    # Series of operations
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    
    # Commit transaction
    conn.commit()
    print("Transaction completed successfully")
    
except IntegrityError as e:
    print(f"Constraint violation: {e}")
    conn.rollback()
    
except OperationalError as e:
    print(f"Database operation failed: {e}")
    conn.rollback()
    
except Exception as e:
    print(f"Unexpected error: {e}")
    conn.rollback()
    
finally:
    conn.autocommit(True)
    cursor.close()
    conn.close()

import cymysql
from cymysql import MySQLError, OperationalError, ProgrammingError
import logging
import time

class DatabaseManager:
    def __init__(self, **conn_params):
        self.conn_params = conn_params
        self.conn = None
        self.max_retries = 3
        self.retry_delay = 1
    
    def connect(self):
        """Connect with retry logic."""
        for attempt in range(self.max_retries):
            try:
                self.conn = cymysql.connect(**self.conn_params)
                logging.info("Database connection established")
                return
            except OperationalError as e:
                logging.warning(f"Connection attempt {attempt + 1} failed: {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay)
                else:
                    raise
    
    def execute_query(self, query, params=None):
        """Execute query with error handling."""
        if not self.conn:
            self.connect()
        
        cursor = None
        try:
            cursor = self.conn.cursor()
            cursor.execute(query, params)
            
            if query.strip().upper().startswith('SELECT'):
                return cursor.fetchall()
            else:
                self.conn.commit()
                return cursor.rowcount
                
        except ProgrammingError as e:
            logging.error(f"SQL programming error: {e}")
            raise
        except OperationalError as e:
            logging.error(f"Database operational error: {e}")
            # Try to reconnect for connection issues
            if "connection" in str(e).lower():
                self.conn = None
                self.connect()
            raise
        except MySQLError as e:
            logging.error(f"MySQL error: {e}")
            self.conn.rollback()
            raise
        finally:
            if cursor:
                cursor.close()
    
    def close(self):
        """Close connection safely."""
        if self.conn:
            try:
                self.conn.close()
                logging.info("Database connection closed")
            except Exception as e:
                logging.warning(f"Error closing connection: {e}")

# Usage
db = DatabaseManager(host='localhost', user='root', password='', db='test')

try:
    result = db.execute_query("SELECT COUNT(*) FROM users")
    print(f"User count: {result[0][0]}")
    
    db.execute_query(
        "INSERT INTO users (name, email) VALUES (%s, %s)",
        ('John Doe', 'john@example.com')
    )
    
except MySQLError as e:
    print(f"Database error occurred: {e}")
finally:
    db.close()

import asyncio
import cymysql.aio
from cymysql import OperationalError, ProgrammingError

async def async_error_handling_example():
    conn = None
    try:
        conn = await cymysql.aio.connect(
            host='localhost',
            user='root',
            password='',
            db='test'
        )
        
        async with conn.cursor() as cursor:
            try:
                await cursor.execute("SELECT * FROM nonexistent_table")
                result = await cursor.fetchall()
            except ProgrammingError as e:
                print(f"Table not found: {e}")
                
            try:
                # Connection timeout test
                await cursor.execute("SELECT SLEEP(30)")
            except OperationalError as e:
                print(f"Query timeout: {e}")
                
    except OperationalError as e:
        print(f"Connection error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")
    finally:
        if conn:
            conn.close()

asyncio.run(async_error_handling_example())

import cymysql
from cymysql.constants import ER

# Common MySQL error codes
ERROR_CODES = {
    ER.ACCESS_DENIED_ERROR: "Access denied",
    ER.BAD_DB_ERROR: "Unknown database",
    ER.BAD_TABLE_ERROR: "Unknown table",
    ER.DUP_ENTRY: "Duplicate entry",
    ER.NO_SUCH_TABLE: "Table doesn't exist",
    ER.SYNTAX_ERROR: "SQL syntax error",
    ER.LOCK_DEADLOCK: "Deadlock found"
}

def handle_mysql_error(e):
    """Handle MySQL error with specific error code logic."""
    if hasattr(e, 'errno') and e.errno in ERROR_CODES:
        print(f"Known error: {ERROR_CODES[e.errno]}")
        
        # Specific handling for different error types
        if e.errno == ER.LOCK_DEADLOCK:
            print("Deadlock detected - retry transaction")
        elif e.errno == ER.DUP_ENTRY:
            print("Duplicate key - check unique constraints")
        elif e.errno == ER.ACCESS_DENIED_ERROR:
            print("Check user permissions")
    else:
        print(f"Unknown MySQL error: {e}")

# Usage in exception handler
try:
    # Database operation
    pass
except cymysql.MySQLError as e:
    handle_mysql_error(e)