Error Handling in psycopg2

Exception Hierarchy

psycopg2.Error (base)
├── DatabaseError
│   ├── DataError           - invalid data
│   ├── OperationalError    - connection issues
│   ├── IntegrityError      - constraint violations
│   ├── InternalError       - internal postgres error
│   ├── ProgrammingError    - SQL syntax errors
│   └── NotSupportedError   - unsupported operation
├── InterfaceError          - interface problems
└── Warning

Handling Specific Errors

import psycopg2
from psycopg2 import errors

try:
    cur.execute(
        "INSERT INTO users (email) VALUES (%s)",
        ("duplicate@email.com",)
    )
    conn.commit()

except errors.UniqueViolation as e:
    conn.rollback()
    print(f"Duplicate entry: {e.diag.message_detail}")

except errors.ForeignKeyViolation as e:
    conn.rollback()
    print(f"Referenced record doesn't exist: {e}")

except errors.CheckViolation as e:
    conn.rollback()
    print(f"Check constraint failed: {e}")

except psycopg2.IntegrityError as e:
    conn.rollback()
    print(f"Data integrity error: {e}")

except psycopg2.OperationalError as e:
    print(f"Connection/operational error: {e}")
    # May need to reconnect

except psycopg2.ProgrammingError as e:
    print(f"SQL error: {e}")

except psycopg2.Error as e:
    conn.rollback()
    print(f"Database error: {e}")

Error Details

try:
    cur.execute("SELECT * FROM nonexistent_table")
except psycopg2.Error as e:
    # Access error details
    print(f"Error code: {e.pgcode}")
    print(f"Error message: {e.pgerror}")

    # Diagnostic info
    if e.diag:
        print(f"Severity: {e.diag.severity}")
        print(f"Detail: {e.diag.message_detail}")
        print(f"Hint: {e.diag.message_hint}")
        print(f"Position: {e.diag.statement_position}")

Retry Pattern

import time
from psycopg2 import OperationalError

def execute_with_retry(conn, query, params=None, max_retries=3):
    """Execute query with automatic retry on transient errors"""
    for attempt in range(max_retries):
        try:
            with conn.cursor() as cur:
                cur.execute(query, params)
                conn.commit()
                return cur.fetchall() if cur.description else None

        except OperationalError as e:
            conn.rollback()
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # Exponential backoff
                print(f"Retry {attempt + 1} in {wait}s: {e}")
                time.sleep(wait)
            else:
                raise

Database Error Handling

Database Error Handling

Error Handling in psycopg2

Exception Hierarchy

Handling Specific Errors

Error Details

Retry Pattern