tessl/pypi-py-postgresql
PostgreSQL driver and tools library providing PG-API and DB-API 2.0 interfaces for Python.
—
Pending
exception-handling.mddocs/
Exception Handling
Complete exception hierarchy mapping PostgreSQL error codes to specific Python exception classes with detailed error information and SQL state codes.
Capabilities
Base Exception Classes
Core exception classes providing the foundation for PostgreSQL error handling.
class Exception(Exception):
"""
Base class for all py-postgresql exceptions.
"""
@property
def code():
"""
Get PostgreSQL error/state code.
Returns:
str: PostgreSQL SQLSTATE code (5 characters)
"""
@property
def message():
"""
Get error message.
Returns:
str: Descriptive error message
"""
class LoadError(Exception):
"""
Raised when library loading fails (C extensions, shared libraries).
"""
class Disconnection(Exception):
"""
Raised when connection is unexpectedly terminated.
"""Core Error Categories
Primary error categories matching PostgreSQL error classes.
class Error(Exception):
"""
Base class for all database operation errors.
"""
class DriverError(Error):
"""
Errors originating from the driver implementation rather than PostgreSQL.
"""
class ConnectionError(Error):
"""
Errors related to database connection establishment or maintenance.
"""
class TransactionError(Error):
"""
Errors related to transaction management and control.
"""
class QueryError(Error):
"""
Errors related to query execution and statement processing.
"""
class AuthenticationSpecificationError(Error):
"""
Errors in authentication specification or credentials.
"""SQL State Error Classes
Specific error classes mapped to PostgreSQL SQL state codes for precise error handling.
# Class 08 - Connection Exception
class SEARVError(Error):
"""Connection exceptions (SQL state class 08)."""
# Class 23 - Integrity Constraint Violation
class ICVError(Error):
"""Integrity constraint violations (SQL state class 23)."""
# Class 22 - Data Exception
class DataError(Error):
"""Data exceptions (SQL state class 22)."""
# Class XX - Internal Error
class InternalError(Error):
"""Internal PostgreSQL errors (SQL state class XX)."""
# Class 42 - Syntax Error or Access Rule Violation
class ProgrammingError(Error):
"""Programming errors - syntax errors, access violations (SQL state class 42)."""
# Class P0 - PL/pgSQL Error
class PLPGSQLError(Error):
"""PL/pgSQL procedure errors (SQL state class P0)."""
# Class 53 - Insufficient Resources
class InsufficientResourcesError(Error):
"""System resource errors (SQL state class 53)."""
# Class 54 - Program Limit Exceeded
class ProgramLimitExceededError(Error):
"""Program limit exceeded errors (SQL state class 54)."""
# Class 55 - Object Not In Prerequisite State
class ObjectNotInPrerequisiteStateError(Error):
"""Object state errors (SQL state class 55)."""
# Class 57 - Operator Intervention
class OperatorInterventionError(Error):
"""Operator intervention errors (SQL state class 57)."""
# Class 58 - System Error
class SystemError(Error):
"""System errors (SQL state class 58)."""Warning Categories
Warning classes for non-fatal conditions that may require attention.
class Warning(Exception):
"""
Base class for all warnings.
"""
class DriverWarning(Warning):
"""
Warnings from driver implementation.
"""
class DeprecationWarning(Warning):
"""
Warnings about deprecated functionality.
"""
class OptimizationWarning(Warning):
"""
Warnings about performance or optimization issues.
"""
class SecurityWarning(Warning):
"""
Warnings about security-related issues.
"""
class UnsupportedWarning(Warning):
"""
Warnings about unsupported operations or features.
"""Exception Lookup Functions
Functions for dynamically resolving exception classes based on PostgreSQL error codes.
def ErrorLookup(state_code):
"""
Get appropriate exception class for PostgreSQL SQL state code.
Parameters:
- state_code (str): 5-character PostgreSQL SQLSTATE code
Returns:
type: Exception class corresponding to the error code
Example:
- ErrorLookup('23505') returns ICVError (unique violation)
- ErrorLookup('42P01') returns ProgrammingError (undefined table)
"""
def WarningLookup(state_code):
"""
Get appropriate warning class for PostgreSQL SQL state code.
Parameters:
- state_code (str): 5-character PostgreSQL SQLSTATE code
Returns:
type: Warning class corresponding to the warning code
"""Usage Examples
Basic Exception Handling
import postgresql
import postgresql.exceptions as pg_exc
try:
db = postgresql.open('pq://user:wrongpass@localhost/mydb')
except pg_exc.AuthenticationSpecificationError as e:
print(f"Authentication failed: {e}")
print(f"Error code: {e.code}")
except pg_exc.ConnectionError as e:
print(f"Connection failed: {e}")
print(f"Error code: {e.code}")
except pg_exc.Error as e:
print(f"Database error: {e}")
print(f"Error code: {e.code}")Query Exception Handling
import postgresql
import postgresql.exceptions as pg_exc
db = postgresql.open('pq://user:pass@localhost/mydb')
try:
# This may raise various exceptions
stmt = db.prepare("SELECT * FROM nonexistent_table WHERE id = $1")
result = stmt.first(123)
except pg_exc.ProgrammingError as e:
print(f"SQL programming error: {e}")
if e.code == '42P01': # undefined_table
print("Table does not exist")
elif e.code == '42703': # undefined_column
print("Column does not exist")
else:
print(f"Other programming error: {e.code}")
except pg_exc.DataError as e:
print(f"Data error: {e}")
if e.code == '22P02': # invalid_text_representation
print("Invalid input format for type")
elif e.code == '22003': # numeric_value_out_of_range
print("Numeric value out of range")
else:
print(f"Other data error: {e.code}")
except pg_exc.Error as e:
print(f"General database error: {e}")
print(f"SQL state: {e.code}")Transaction Exception Handling
import postgresql
import postgresql.exceptions as pg_exc
db = postgresql.open('pq://user:pass@localhost/mydb')
try:
with db.xact():
# Insert user
insert_user = db.prepare("INSERT INTO users (email, name) VALUES ($1, $2)")
insert_user("john@example.com", "John Doe")
# Insert duplicate email (assuming unique constraint)
insert_user("john@example.com", "Jane Doe") # This will fail
# This won't be reached due to exception
print("Both users inserted successfully")
except pg_exc.ICVError as e:
print(f"Integrity constraint violation: {e}")
if e.code == '23505': # unique_violation
print("Duplicate value violates unique constraint")
elif e.code == '23503': # foreign_key_violation
print("Foreign key constraint violation")
elif e.code == '23514': # check_violation
print("Check constraint violation")
else:
print(f"Other integrity error: {e.code}")
except pg_exc.TransactionError as e:
print(f"Transaction error: {e}")
print(f"Transaction was rolled back")Dynamic Exception Resolution
import postgresql
import postgresql.exceptions as pg_exc
db = postgresql.open('pq://user:pass@localhost/mydb')
def handle_database_error(exception):
"""Handle database errors based on SQL state code."""
if hasattr(exception, 'code') and exception.code:
# Get specific exception class for this error code
error_class = pg_exc.ErrorLookup(exception.code)
print(f"Error type: {error_class.__name__}")
print(f"SQL state: {exception.code}")
print(f"Message: {exception}")
# Handle specific error categories
if issubclass(error_class, pg_exc.ICVError):
handle_integrity_error(exception)
elif issubclass(error_class, pg_exc.ProgrammingError):
handle_programming_error(exception)
elif issubclass(error_class, pg_exc.DataError):
handle_data_error(exception)
else:
print("General database error")
else:
print(f"Unknown error: {exception}")
def handle_integrity_error(e):
"""Handle integrity constraint violations."""
constraints = {
'23505': 'Unique constraint violation - duplicate value',
'23503': 'Foreign key constraint violation - referenced record not found',
'23514': 'Check constraint violation - invalid value',
'23502': 'Not null constraint violation - required field is null'
}
print(f"Integrity issue: {constraints.get(e.code, 'Unknown constraint violation')}")
def handle_programming_error(e):
"""Handle SQL programming errors."""
errors = {
'42P01': 'Table does not exist',
'42703': 'Column does not exist',
'42883': 'Function does not exist',
'42P07': 'Object already exists',
'42601': 'Syntax error'
}
print(f"Programming issue: {errors.get(e.code, 'SQL programming error')}")
def handle_data_error(e):
"""Handle data processing errors."""
errors = {
'22P02': 'Invalid input format for type',
'22003': 'Numeric value out of range',
'22007': 'Invalid datetime format',
'22012': 'Division by zero'
}
print(f"Data issue: {errors.get(e.code, 'Data processing error')}")
# Use the error handler
try:
stmt = db.prepare("INSERT INTO users (id, email) VALUES ($1, $2)")
stmt(999999999999999999999, "invalid-email") # Will cause data error
except pg_exc.Error as e:
handle_database_error(e)Exception Hierarchy Usage
import postgresql
import postgresql.exceptions as pg_exc
def robust_database_operation(db, query, *params):
"""Execute database operation with comprehensive error handling."""
try:
stmt = db.prepare(query)
return stmt(*params)
except pg_exc.AuthenticationSpecificationError:
print("Authentication problem - check credentials")
raise
except pg_exc.ConnectionError:
print("Connection problem - check network/server")
raise
except pg_exc.ICVError as e:
print(f"Data integrity issue: {e}")
# Don't re-raise - handle gracefully
return None
except pg_exc.ProgrammingError as e:
print(f"SQL programming error: {e}")
# Log and re-raise for developer attention
import logging
logging.error(f"SQL error in query '{query}': {e}")
raise
except pg_exc.DataError as e:
print(f"Data format error: {e}")
# Return None for data errors
return None
except pg_exc.TransactionError as e:
print(f"Transaction error: {e}")
# Transaction errors should be re-raised
raise
except pg_exc.DriverError as e:
print(f"Driver error: {e}")
# Driver errors are usually fatal
raise
except pg_exc.Error as e:
print(f"General database error: {e}")
# Catch-all for other database errors
raise
# Usage
db = postgresql.open('pq://user:pass@localhost/mydb')
result = robust_database_operation(
db,
"SELECT * FROM users WHERE id = $1",
123
)
if result:
print(f"Found {len(result)} users")
else:
print("Query failed or returned no results")Warning Handling
import postgresql
import postgresql.exceptions as pg_exc
import warnings
# Configure warning handling
warnings.filterwarnings('always', category=pg_exc.DriverWarning)
warnings.filterwarnings('always', category=pg_exc.OptimizationWarning)
db = postgresql.open('pq://user:pass@localhost/mydb')
# Function that might generate warnings
def execute_with_warnings(db, query):
try:
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
result = db.query(query)
# Check for warnings
for warning in w:
if issubclass(warning.category, pg_exc.DriverWarning):
print(f"Driver warning: {warning.message}")
elif issubclass(warning.category, pg_exc.OptimizationWarning):
print(f"Performance warning: {warning.message}")
elif issubclass(warning.category, pg_exc.DeprecationWarning):
print(f"Deprecation warning: {warning.message}")
else:
print(f"Other warning: {warning.message}")
return result
except pg_exc.Error as e:
print(f"Error executing query: {e}")
return None
# Example usage
result = execute_with_warnings(db, "SELECT * FROM large_table LIMIT 1000000")Install with Tessl CLI
npx tessl i tessl/pypi-py-postgresql