Error Handling and Exceptions

class Error(Exception):
    """
    Base exception class for all MySQL errors.
    
    All MySQL-specific exceptions inherit from this base class,
    providing common attributes and methods for error handling.
    """
    
    def __init__(self, msg=None, errno=None, values=None, sqlstate=None):
        """
        Initialize error with message and optional details.
        
        Parameters:
        - msg (str): Error message
        - errno (int): MySQL error number
        - values (tuple): Message format values
        - sqlstate (str): SQL state code
        """
    
    @property
    def errno(self):
        """MySQL error number"""
    
    @property
    def sqlstate(self):
        """SQL state code"""
    
    @property
    def msg(self):
        """Error message"""

class Warning(Error):
    """
    Exception for important warnings.
    
    Raised for significant conditions that are not errors but should
    be brought to the application's attention.
    """

class InterfaceError(Error):
    """
    Interface-related errors.
    
    Raised for errors related to the database interface rather than
    the database itself. Examples: connection parameter errors,
    programming interface misuse.
    """

class DatabaseError(Error):
    """
    Database-related errors.
    
    Base class for errors that are related to the database operation
    itself rather than interface issues.
    """

class DataError(DatabaseError):
    """
    Data processing errors.
    
    Raised for errors due to problems with the processed data:
    numeric value out of range, division by zero, invalid date, etc.
    """

class OperationalError(DatabaseError):
    """
    Database operation errors.
    
    Raised for errors related to database operation and not necessarily
    under the control of the programmer: connection lost, database name
    not found, transaction could not be processed, memory allocation error.
    """

class IntegrityError(DatabaseError):
    """
    Relational integrity violations.
    
    Raised when the relational integrity of the database is affected:
    foreign key check fails, duplicate key, constraint violations.
    """

class InternalError(DatabaseError):
    """
    Internal database errors.
    
    Raised when the database encounters an internal error: cursor is
    not valid anymore, transaction is out of sync, etc.
    """

class ProgrammingError(DatabaseError):
    """
    Programming errors.
    
    Raised for programming errors: table not found, error in SQL syntax,
    wrong number of parameters specified, etc.
    """

class NotSupportedError(DatabaseError):
    """
    Unsupported feature errors.
    
    Raised when using a database method or feature that is not supported
    by the database version or configuration.
    """

class PoolError(Error):
    """
    Connection pooling errors.
    
    Raised for errors specific to connection pool operations:
    pool full, invalid pool configuration, pool connection errors.
    """

def custom_error_exception(error=None, exception=None):
    """
    Define custom exceptions for MySQL server errors.
    
    Allows mapping specific MySQL error codes to custom exception classes
    for more granular error handling.
    
    Parameters:
    - error (dict): Error code to exception class mapping
    - exception (Exception): Custom exception class
    
    Returns:
    dict: Current error mappings
    
    Example:
    # Map specific error to custom exception
    custom_error_exception({1050: MyTableExistsError})
    
    # Define custom exception for error code
    class MyDuplicateKeyError(IntegrityError):
        pass
    
    custom_error_exception({1062: MyDuplicateKeyError})
    """

import mysql.connector
from mysql.connector import Error, IntegrityError, ProgrammingError

# Custom exception for duplicate keys
class DuplicateKeyError(IntegrityError):
    def __init__(self, msg=None, errno=None, values=None, sqlstate=None):
        super().__init__(msg, errno, values, sqlstate)
        self.duplicate_key = True

# Register custom exception
mysql.connector.custom_error_exception({1062: DuplicateKeyError})

try:
    connection = mysql.connector.connect(
        user='myuser',
        password='mypassword',
        host='localhost',
        database='mydatabase'
    )
    
    cursor = connection.cursor()
    cursor.execute("INSERT INTO users (email) VALUES (%s)", ("john@example.com",))
    
except DuplicateKeyError as err:
    print(f"Duplicate key error: {err}")
    print(f"Error code: {err.errno}")
    print(f"SQL state: {err.sqlstate}")
    
except IntegrityError as err:
    print(f"Integrity constraint violation: {err}")
    
except ProgrammingError as err:
    print(f"Programming error: {err}")
    
except Error as err:
    print(f"MySQL error: {err}")
    
finally:
    if 'connection' in locals() and connection.is_connected():
        cursor.close()
        connection.close()

class errorcode:
    """
    MySQL server error codes.
    
    Constants for common MySQL error conditions to enable
    specific error handling based on error codes.
    """
    
    # Connection errors
    CR_CONNECTION_ERROR: int = 2003
    CR_CONN_HOST_ERROR: int = 2005
    CR_SERVER_GONE_ERROR: int = 2006
    CR_SERVER_LOST: int = 2013
    
    # Authentication errors  
    ER_ACCESS_DENIED_ERROR: int = 1045
    ER_BAD_DB_ERROR: int = 1049
    
    # Data integrity errors
    ER_DUP_ENTRY: int = 1062
    ER_NO_REFERENCED_ROW: int = 1216
    ER_ROW_IS_REFERENCED: int = 1217
    ER_NO_REFERENCED_ROW_2: int = 1452
    ER_ROW_IS_REFERENCED_2: int = 1451
    
    # Schema errors
    ER_BAD_TABLE_ERROR: int = 1051
    ER_NO_SUCH_TABLE: int = 1146
    ER_TABLE_EXISTS_ERROR: int = 1050
    ER_BAD_FIELD_ERROR: int = 1054
    ER_DUP_FIELDNAME: int = 1060
    
    # Syntax errors
    ER_PARSE_ERROR: int = 1064
    ER_WRONG_VALUE_COUNT_ON_ROW: int = 1136
    
    # Lock errors
    ER_LOCK_WAIT_TIMEOUT: int = 1205
    ER_LOCK_DEADLOCK: int = 1213
    
    # Storage errors
    ER_DISK_FULL: int = 1021
    ER_OUT_OF_RESOURCES: int = 1041

import mysql.connector
from mysql.connector import errorcode

try:
    connection = mysql.connector.connect(**config)
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM nonexistent_table")
    
except mysql.connector.Error as err:
    if err.errno == errorcode.ER_NO_SUCH_TABLE:
        print("Table doesn't exist")
    elif err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
        print("Access denied")
    elif err.errno == errorcode.ER_BAD_DB_ERROR:
        print("Database doesn't exist")
    else:
        print(f"Error {err.errno}: {err}")

class MySQLConnection:
    """Connection with context manager support"""
    
    def __enter__(self):
        """Enter context manager"""
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        """Exit context manager and close connection"""
        self.close()

class MySQLCursor:
    """Cursor with context manager support"""
    
    def __enter__(self):
        """Enter context manager"""
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        """Exit context manager and close cursor"""
        self.close()

import mysql.connector

# Connection context manager
with mysql.connector.connect(**config) as connection:
    with connection.cursor() as cursor:
        cursor.execute("SELECT * FROM users")
        results = cursor.fetchall()
        
        for row in results:
            print(row)
    # Cursor automatically closed
# Connection automatically closed

# Handle exceptions with context manager
try:
    with mysql.connector.connect(**config) as connection:
        with connection.cursor() as cursor:
            cursor.execute("INSERT INTO users (email) VALUES (%s)", ("test@example.com",))
            connection.commit()
            
except mysql.connector.IntegrityError:
    print("Integrity constraint violation")
except mysql.connector.Error as err:
    print(f"Database error: {err}")

# mysqlx module exceptions (same hierarchy as mysql.connector)
class Error(Exception):
    """Base exception for X DevAPI operations"""

class Warning(Error):
    """Warning exception for X DevAPI"""

class InterfaceError(Error):
    """Interface error for X DevAPI"""

class DatabaseError(Error):
    """Database error for X DevAPI"""

class DataError(DatabaseError):
    """Data processing error for X DevAPI"""

class OperationalError(DatabaseError):
    """Operational error for X DevAPI"""

class IntegrityError(DatabaseError):
    """Integrity constraint violation for X DevAPI"""

class InternalError(DatabaseError):
    """Internal error for X DevAPI"""

class ProgrammingError(DatabaseError):
    """Programming error for X DevAPI"""

class NotSupportedError(DatabaseError):
    """Unsupported feature error for X DevAPI"""

import mysqlx

try:
    session = mysqlx.get_session({
        'host': 'localhost',
        'port': 33060,
        'user': 'myuser',
        'password': 'mypassword'
    })
    
    schema = session.get_schema('mydb')
    collection = schema.get_collection('users')
    
    # This might raise IntegrityError if document violates constraints
    collection.add({
        'email': 'duplicate@example.com',
        'user_id': 123
    }).execute()
    
except mysqlx.InterfaceError as err:
    print(f"Connection or interface error: {err}")
    
except mysqlx.IntegrityError as err:
    print(f"Document violates constraints: {err}")
    
except mysqlx.OperationalError as err:
    print(f"Database operation failed: {err}")
    
except mysqlx.Error as err:
    print(f"X DevAPI error: {err}")
    
finally:
    if 'session' in locals():
        session.close()

import logging

# Enable MySQL Connector logging
logger = logging.getLogger('mysql.connector')
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)

import logging
import mysql.connector

# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('mysql.connector')

try:
    connection = mysql.connector.connect(
        user='myuser',
        password='mypassword',
        host='localhost',
        database='mydatabase'
    )
    
    cursor = connection.cursor()
    cursor.execute("SELECT COUNT(*) FROM users")
    result = cursor.fetchone()
    print(f"User count: {result[0]}")
    
except mysql.connector.Error as err:
    logger.error(f"Database error: {err}")
    
finally:
    if 'connection' in locals() and connection.is_connected():
        cursor.close()
        connection.close()

def custom_error_exception(error=None, exception=None):
    """
    Define custom exceptions for MySQL server errors.
    
    This function defines custom exceptions for MySQL server errors and
    returns the current set customizations. If error is a MySQL Server 
    error number, then you have to pass also the exception class.
    
    Parameters:
    - error (int|dict): MySQL error number or dictionary mapping errors to exceptions
    - exception (Exception): Exception class to raise for the error
    
    Returns:
    dict: Current set of custom error mappings
    
    Raises:
    ValueError: Invalid error number or exception type
    
    Examples:
    # Map specific error to custom exception
    mysql.connector.custom_error_exception(1028, mysql.connector.DatabaseError)
    
    # Map multiple errors using dictionary
    mysql.connector.custom_error_exception({
        1028: mysql.connector.DatabaseError,
        1029: mysql.connector.OperationalError,
    })
    
    # Reset all customizations
    mysql.connector.custom_error_exception({})
    """

def read_option_files(*args, **kwargs):
    """
    Read MySQL option files for connection configuration.
    
    Parameters:
    - *args: Option file paths
    - **kwargs: Configuration options including 'option_files' key
    
    Returns:
    dict: Configuration dictionary with values from option files
    
    Example:
    config = mysql.connector.read_option_files(
        option_files=['~/.my.cnf', '/etc/mysql/my.cnf']
    )
    connection = mysql.connector.connect(**config)
    """

ErrorMapping = dict[int, type[Exception]]

ExceptionInfo = {
    'errno': int,
    'sqlstate': str, 
    'msg': str,
    'values': tuple
}