Error Handling

class Warning(Exception):
    """
    Not used by phoenixdb, only defined for DB API 2.0 compatibility.
    """

class Error(Exception):
    """
    Base class for all other error exceptions.
    Can be used to catch all errors with single except statement.
    """

    def __init__(self, message, code=None, sqlstate=None, cause=None): ...

    @property
    def message(self):
        """Error message string."""

    @property
    def code(self):
        """Phoenix/database error code (integer or None)."""

    @property
    def sqlstate(self):
        """SQL state code string (SQLSTATE standard)."""

    @property
    def cause(self):
        """Underlying cause exception or None."""

class InterfaceError(Error):
    """
    Errors related to the database interface rather than the database itself.
    Examples: connection problems, parameter issues, API usage errors.
    """

class DatabaseError(Error):
    """
    Base class for errors related to the database.
    All database-specific errors inherit from this.
    """

class DataError(DatabaseError):
    """
    Errors due to problems with processed data.
    Examples: division by zero, numeric value out of range, type conversion errors.
    """

class OperationalError(DatabaseError):
    """
    Errors related to database operation, not necessarily under programmer control.
    Examples: disconnect, memory allocation error, transaction processing failure.
    """

class IntegrityError(DatabaseError):
    """
    Errors when relational integrity of database is affected.
    Examples: foreign key check fails, unique constraint violation.
    """

class InternalError(DatabaseError):
    """
    Errors when database encounters an internal problem.
    Examples: cursor not valid, transaction out of sync.
    """

class ProgrammingError(DatabaseError):
    """
    Programming errors.
    Examples: table not found, SQL syntax error, wrong parameter count.
    """

class NotSupportedError(DatabaseError):
    """
    Errors when using API not supported by database.
    Examples: unsupported SQL feature, method not implemented.
    """

import phoenixdb

try:
    conn = phoenixdb.connect('http://localhost:8765/')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()
except phoenixdb.Error as e:
    print(f"Database error: {e.message}")
    if e.code:
        print(f"Error code: {e.code}")
    if e.sqlstate:
        print(f"SQL state: {e.sqlstate}")
    if e.cause:
        print(f"Caused by: {e.cause}")

import phoenixdb

try:
    conn = phoenixdb.connect('http://invalid-url:8765/')
except phoenixdb.InterfaceError as e:
    print(f"Connection interface error: {e.message}")
except phoenixdb.OperationalError as e:
    print(f"Connection failed: {e.message}")

try:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM nonexistent_table")
except phoenixdb.ProgrammingError as e:
    print(f"SQL programming error: {e.message}")
    print(f"SQL state: {e.sqlstate}")

try:
    cursor.execute("INSERT INTO users VALUES (?, ?)", (1,))  # Wrong parameter count
except phoenixdb.ProgrammingError as e:
    print(f"Parameter error: {e.message}")

try:
    cursor.execute("INSERT INTO users (id) VALUES (?)", ("not_a_number",))
except phoenixdb.DataError as e:
    print(f"Data type error: {e.message}")

import phoenixdb

try:
    # Invalid URL format
    conn = phoenixdb.connect('invalid://url')
except phoenixdb.InterfaceError as e:
    print(f"Invalid connection parameters: {e.message}")

try:
    # Server not available
    conn = phoenixdb.connect('http://nonexistent-server:8765/')
except phoenixdb.OperationalError as e:
    print(f"Cannot connect to server: {e.message}")

try:
    # Authentication failure
    conn = phoenixdb.connect('http://localhost:8765/',
                           authentication='BASIC',
                           avatica_user='wrong_user',
                           avatica_password='wrong_password')
except phoenixdb.OperationalError as e:
    print(f"Authentication failed: {e.message}")

import phoenixdb

conn = phoenixdb.connect('http://localhost:8765/')
cursor = conn.cursor()

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

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

try:
    # Column doesn't exist
    cursor.execute("SELECT nonexistent_column FROM users")
except phoenixdb.ProgrammingError as e:
    print(f"Column not found: {e.message}")

try:
    # Division by zero
    cursor.execute("SELECT 1/0")
except phoenixdb.DataError as e:
    print(f"Data processing error: {e.message}")

import phoenixdb

conn = phoenixdb.connect('http://localhost:8765/')
cursor = conn.cursor()

try:
    # Wrong number of parameters
    cursor.execute("INSERT INTO users (id, name) VALUES (?, ?)", (1,))  # Missing parameter
except phoenixdb.ProgrammingError as e:
    print(f"Parameter count mismatch: {e.message}")

try:
    # Invalid parameter type for array
    cursor.execute("INSERT INTO test_array (id, numbers) VALUES (?, ?)", (1, "not_an_array"))
except phoenixdb.ProgrammingError as e:
    print(f"Invalid parameter type: {e.message}")

import phoenixdb

conn = phoenixdb.connect('http://localhost:8765/')
cursor = conn.cursor()

try:
    # Create table with primary key
    cursor.execute("CREATE TABLE test_pk (id INTEGER PRIMARY KEY, name VARCHAR)")

    # Insert first row
    cursor.execute("INSERT INTO test_pk VALUES (1, 'first')")

    # Try to insert duplicate primary key
    cursor.execute("INSERT INTO test_pk VALUES (1, 'duplicate')")
except phoenixdb.IntegrityError as e:
    print(f"Primary key violation: {e.message}")

import phoenixdb

conn = phoenixdb.connect('http://localhost:8765/')
cursor = conn.cursor()

try:
    # Try to fetch without executing query
    cursor.fetchone()
except phoenixdb.ProgrammingError as e:
    print(f"No select statement executed: {e.message}")

# Close cursor
cursor.close()

try:
    # Try to use closed cursor
    cursor.execute("SELECT 1")
except phoenixdb.ProgrammingError as e:
    print(f"Cursor is closed: {e.message}")

import phoenixdb

conn = phoenixdb.connect('http://localhost:8765/')
cursor = conn.cursor()

# Close connection
conn.close()

try:
    # Try to use closed connection
    cursor.execute("SELECT 1")
except phoenixdb.ProgrammingError as e:
    print(f"Connection is closed: {e.message}")

try:
    # Try to create cursor from closed connection
    new_cursor = conn.cursor()
except phoenixdb.ProgrammingError as e:
    print(f"Connection is closed: {e.message}")

import phoenixdb
import traceback

def handle_database_error(e):
    """Comprehensive error information display."""
    print(f"Error Type: {type(e).__name__}")
    print(f"Message: {e.message}")

    if hasattr(e, 'code') and e.code is not None:
        print(f"Error Code: {e.code}")

    if hasattr(e, 'sqlstate') and e.sqlstate:
        print(f"SQL State: {e.sqlstate}")

    if hasattr(e, 'cause') and e.cause:
        print(f"Underlying Cause: {e.cause}")

    print("Traceback:")
    traceback.print_exc()

try:
    conn = phoenixdb.connect('http://localhost:8765/')
    cursor = conn.cursor()
    cursor.execute("INVALID SQL SYNTAX")
except phoenixdb.Error as e:
    handle_database_error(e)

import phoenixdb
import time

def create_connection_with_retry(url, max_retries=3, delay=1):
    """Create connection with retry logic."""
    for attempt in range(max_retries):
        try:
            return phoenixdb.connect(url)
        except phoenixdb.OperationalError as e:
            if attempt < max_retries - 1:
                print(f"Connection attempt {attempt + 1} failed: {e.message}")
                time.sleep(delay)
                continue
            else:
                print(f"All connection attempts failed")
                raise

# Usage
try:
    conn = create_connection_with_retry('http://localhost:8765/')
except phoenixdb.OperationalError:
    print("Could not establish connection after retries")

import phoenixdb

def execute_with_rollback(conn, statements):
    """Execute statements with automatic rollback on error."""
    cursor = conn.cursor()
    try:
        # Execute all statements
        for sql, params in statements:
            cursor.execute(sql, params)

        # Commit if all successful
        conn.commit()
        print("All statements executed successfully")

    except phoenixdb.DatabaseError as e:
        # Rollback on any database error
        print(f"Error occurred: {e.message}")
        print("Rolling back transaction...")
        conn.rollback()
        raise
    finally:
        cursor.close()

# Usage
statements = [
    ("INSERT INTO users VALUES (?, ?)", (1, 'user1')),
    ("INSERT INTO users VALUES (?, ?)", (2, 'user2')),
    ("UPDATE users SET name = ? WHERE id = ?", ('updated', 1))
]

try:
    execute_with_rollback(conn, statements)
except phoenixdb.DatabaseError:
    print("Transaction failed and was rolled back")