Python Database Error Handling

Database operations fail in predictable ways: networks are unreliable, constraints are violated, credentials are wrong. A production Python application must handle these failures gracefully, not crash with an unhandled exception. This tutorial covers the exception hierarchy, how to catch specific errors, retry strategies, and logging patterns that let you debug failures after they occur.

Understanding database errors transforms your code from brittle to resilient. You'll learn to distinguish between transient errors (retry) and permanent errors (fail immediately), implement exponential backoff, and provide users with meaningful feedback instead of a 500 error.

The Exception Hierarchy

SQLite3, psycopg, and other adapters follow PEP 249 (Python Database API), which defines a standard exception hierarchy. The base class is DatabaseError, with specific subclasses for different failures:

StandardError
    └─ Exception
        └─ DatabaseError
            ├─ IntegrityError (constraint violated: unique, foreign key, check)
            ├─ OperationalError (database server issue, connection lost)
            ├─ ProgrammingError (SQL syntax error, invalid parameter)
            └─ InternalError (database internal error)

Here's what each means:

• IntegrityError: A constraint was violated (unique key, foreign key, NOT NULL). Data is safe; the error is recoverable (user should retry with valid data).
• OperationalError: A transient database problem (connection lost, server down, timeout). Usually recoverable with retry; data may be partially committed.
• ProgrammingError: A bug in your code (SQL syntax error, missing column, wrong parameter count). Not recoverable without code changes.
• InternalError: A serious database problem. Rare; usually requires database administrator intervention.

Here's example code catching these:

import sqlite3

conn = sqlite3.connect('app.db')
cursor = conn.cursor()

try:
    cursor.execute('INSERT INTO users (id, name, email) VALUES (?, ?, ?)',
                   (1, 'Alice', '[email protected]'))
    conn.commit()
    print("User created")

except sqlite3.IntegrityError as e:
    # Constraint violated (e.g., duplicate id or email already exists)
    conn.rollback()
    print(f"User already exists or constraint violated: {e}")

except sqlite3.OperationalError as e:
    # Database issue (connection lost, disk full, etc.)
    conn.rollback()
    print(f"Database is unavailable: {e}. Retry later.")

except sqlite3.ProgrammingError as e:
    # Programming bug (SQL syntax error, missing column, etc.)
    conn.rollback()
    print(f"SQL error (check your code): {e}")

except sqlite3.DatabaseError as e:
    # Catch-all for any database error
    conn.rollback()
    print(f"Unexpected database error: {e}")

finally:
    conn.close()

Note that exception class names differ slightly between adapters:

Adapter	Base Exception	Common Subclasses
sqlite3	sqlite3.DatabaseError	IntegrityError, OperationalError, ProgrammingError
psycopg	psycopg.DatabaseError	IntegrityError, OperationalError, ProgrammingError
MySQLdb	MySQLdb.DatabaseError	IntegrityError, OperationalError, ProgrammingError

To write adapter-agnostic code, catch the generic exceptions:

import sqlite3
import psycopg

def insert_user(adapter, connection, user_id, name, email):
    """
    Insert a user, handling errors generically across adapters.
    """
    try:
        cursor = connection.cursor()
        if adapter == 'sqlite3':
            cursor.execute('INSERT INTO users VALUES (?, ?, ?)',
                           (user_id, name, email))
        else:  # psycopg
            cursor.execute('INSERT INTO users VALUES (%s, %s, %s)',
                           (user_id, name, email))
        connection.commit()
        return True

    except sqlite3.IntegrityError as e:
        connection.rollback()
        print(f"Integrity error: {e}")
        return False

    except sqlite3.OperationalError as e:
        connection.rollback()
        print(f"Operational error: {e}")
        return False

    finally:
        # Don't close connection if it's passed in; let the caller manage it
        pass

Retrying Transient Errors

Some errors are transient—the operation will succeed if you retry. Network timeouts, server restarts, and temporary lock contention are examples. Implement retry logic with exponential backoff:

import sqlite3
import time
import random

def execute_with_retry(cursor, query, params=None, max_retries=3):
    """
    Execute a query with exponential backoff retry on transient errors.
    Raises the exception if max_retries is exceeded or the error is permanent.
    """
    for attempt in range(max_retries):
        try:
            if params:
                cursor.execute(query, params)
            else:
                cursor.execute(query)
            return cursor.fetchall() if query.strip().upper().startswith('SELECT') else None

        except sqlite3.OperationalError as e:
            # Transient error; retry
            if attempt < max_retries - 1:
                # Exponential backoff with jitter
                wait_time = 2 ** attempt + random.uniform(0, 1)
                print(f"Transient error: {e}. Retrying in {wait_time:.1f} seconds...")
                time.sleep(wait_time)
            else:
                raise  # Re-raise on final attempt

        except sqlite3.ProgrammingError as e:
            # Programming error; don't retry
            print(f"Programming error (no retry): {e}")
            raise

        except sqlite3.IntegrityError as e:
            # Integrity error; don't retry (data is invalid)
            print(f"Integrity error (no retry): {e}")
            raise

# Usage
conn = sqlite3.connect('app.db')
cursor = conn.cursor()

try:
    cursor.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)')
    execute_with_retry(
        cursor,
        'INSERT INTO users (name) VALUES (?)',
        ('Alice',)
    )
    conn.commit()
    print("Insert successful after potential retries")

except Exception as e:
    conn.rollback()
    print(f"Insert failed permanently: {e}")

finally:
    conn.close()

The exponential backoff pattern (wait 1s, 2s, 4s, 8s) prevents overwhelming a recovering server. Add jitter (random.uniform(0, 1)) to avoid thundering herd (all clients retrying simultaneously).

Logging Database Errors

Production applications must log errors for debugging. Use Python's logging module:

import sqlite3
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

conn = sqlite3.connect('app.db')
cursor = conn.cursor()

try:
    cursor.execute('SELECT * FROM users WHERE id = ?', (1,))
    user = cursor.fetchone()

except sqlite3.OperationalError as e:
    logger.error(f"Database connection failed: {e}", exc_info=True)
    # exc_info=True includes the full traceback in the log

except sqlite3.ProgrammingError as e:
    logger.error(f"SQL syntax or schema error: {e}", exc_info=True)

except Exception as e:
    logger.exception(f"Unexpected error: {e}")  # equivalent to exc_info=True

finally:
    conn.close()

Always log exc_info=True (or use logger.exception()) to capture the full traceback. This helps you diagnose the error days or weeks later when a user reports a bug.

For web applications (Django, FastAPI), log with context:

import logging
from fastapi import FastAPI, HTTPException
from psycopg_pool import ConnectionPool

app = FastAPI()
logger = logging.getLogger(__name__)
pool = ConnectionPool('postgresql://user:pass@localhost/myapp', min_size=5, max_size=20)

@app.get("/users/{user_id}")
def get_user(user_id: int):
    try:
        with pool.connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT id, name, email FROM users WHERE id = %s', (user_id,))
            user = cursor.fetchone()

        if not user:
            raise HTTPException(status_code=404, detail=f"User {user_id} not found")

        return {"id": user[0], "name": user[1], "email": user[2]}

    except Exception as e:
        # Log with context for debugging
        logger.error(f"Failed to fetch user {user_id}: {e}", exc_info=True, extra={
            'user_id': user_id,
            'timestamp': time.time(),
        })
        raise HTTPException(status_code=500, detail="Database error. Please retry.")

Handling Specific PostgreSQL Errors

PostgreSQL provides detailed error codes. Catch specific errors by code:

import psycopg
from psycopg import errors

conn = psycopg.connect('postgresql://user:pass@localhost/myapp')
cursor = conn.cursor()

try:
    cursor.execute('INSERT INTO users (id, email) VALUES (%s, %s)',
                   (1, '[email protected]'))
    conn.commit()

except errors.UniqueViolation:
    # Specific error: unique constraint violated
    conn.rollback()
    print("Email already exists. Choose a different email.")

except errors.ForeignKeyViolation:
    # Specific error: foreign key constraint violated
    conn.rollback()
    print("Referenced record does not exist.")

except errors.NotNullViolation:
    # Specific error: NOT NULL constraint violated
    conn.rollback()
    print("A required field is missing.")

except psycopg.IntegrityError as e:
    # Catch all integrity errors (parent class)
    conn.rollback()
    print(f"Data validation failed: {e}")

except psycopg.OperationalError as e:
    # Connection or server issue
    conn.rollback()
    print(f"Database unavailable. Retry later. Error: {e}")

finally:
    conn.close()

PostgreSQL error codes include UniqueViolation (23505), ForeignKeyViolation (23503), NotNullViolation (23502), CheckViolation (23514), and many others. The psycopg.errors module provides named exceptions for the most common ones.

Distinguishing Transient vs. Permanent Errors

A key skill is deciding whether to retry:

Error	Type	Retry?	Example Handling
Connection timeout	Transient	Yes	Wait, then retry (backoff)
Network unreachable	Transient	Yes	Wait, then retry (backoff)
Server down	Transient	Yes	Wait, then retry (backoff)
Unique key violation	Permanent	No	Return 409 Conflict to client
Foreign key violation	Permanent	No	Return 400 Bad Request to client
SQL syntax error	Permanent	No	Log and fix code (this is a bug)
Disk full	Permanent (for writes)	No	Alert database administrator

Use this decision tree:

def should_retry(exception):
    """
    Return True if the error is transient and should be retried.
    """
    transient_errors = [
        'timeout',
        'connection refused',
        'connection reset',
        'temporary failure',
        'server is down',
    ]
    
    error_msg = str(exception).lower()
    return any(err_phrase in error_msg for err_phrase in transient_errors)

# Usage
for attempt in range(3):
    try:
        # Execute query
        break
    except Exception as e:
        if should_retry(e):
            time.sleep(2 ** attempt)
        else:
            raise

Key Takeaways

• Database exceptions follow PEP 249: IntegrityError (constraint), OperationalError (transient), ProgrammingError (bug).
• Distinguish between transient errors (retry with backoff) and permanent errors (fail immediately).
• Always call rollback() if an error occurs during a transaction.
• Log all errors with exc_info=True to capture full tracebacks for debugging.
• Use exponential backoff with jitter when retrying to avoid overwhelming the database.
• Web applications should return meaningful HTTP status codes (409 for constraint violation, 500 for database unavailable).

Frequently Asked Questions

Should I catch all exceptions or just database exceptions?

Catch database-specific exceptions in database code, then let the caller (e.g., a web framework) catch broader exceptions. This separates concerns and makes error handling explicit.

How many times should I retry?

Start with 3 retries (exponential backoff: 1s, 2s, 4s = 7 seconds total). If that's too slow, reduce to 2 retries. Too many retries waste time; too few give up prematurely.

Is it safe to retry after a constraint violation?

No. IntegrityError means the data is invalid; retrying won't help. Return an error to the user instead.

How do I log sensitive data safely?

Avoid logging passwords, API keys, or PII. Use parameterized queries so sensitive values aren't in the SQL string—they're hidden from logs. If you must log them, redact or encrypt.

# WRONG: Password visible in log
logger.info(f"Connecting to user@password:localhost/db")

# RIGHT: Password is hidden
logger.info(f"Connecting to user@***:localhost/db")

Can I use try-except to recover from all errors?

No. Some errors are unrecoverable (disk full, out of memory). Try-except lets you handle errors gracefully, but not all errors can be recovered.

Further Reading

• PEP 249 – Python Database API 2.0 Exceptions
• PostgreSQL Error Codes
• Python logging Documentation
• Exponential Backoff And Jitter