PostgreSQL Constraints & Validation Best Practices¶

Core Principles¶

Data Integrity: Enforce business rules at the database level
Performance: Design constraints for optimal performance
Validation: Implement comprehensive data validation
Error Handling: Provide clear constraint violation messages
Maintenance: Design constraints for long-term maintainability

Primary Key Constraints¶

Primary Key Design¶

-- Create tables with appropriate primary keys
CREATE TABLE users (
    id SERIAL PRIMARY KEY,                    -- Auto-incrementing primary key
    username VARCHAR(50) UNIQUE NOT NULL,    -- Natural key with uniqueness
    email VARCHAR(255) UNIQUE NOT NULL,      -- Natural key with uniqueness
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Composite primary keys for junction tables
CREATE TABLE user_roles (
    user_id INTEGER NOT NULL,
    role_id INTEGER NOT NULL,
    assigned_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
    assigned_by INTEGER,
    PRIMARY KEY (user_id, role_id)            -- Composite primary key
);

-- UUID primary keys for distributed systems
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

CREATE TABLE distributed_entities (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    entity_type VARCHAR(50) NOT NULL,
    data JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Natural primary keys for reference data
CREATE TABLE countries (
    country_code CHAR(2) PRIMARY KEY,         -- ISO country code as natural key
    country_name VARCHAR(100) NOT NULL,
    currency_code CHAR(3) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

Primary Key Performance¶

-- Create indexes for primary key performance
CREATE INDEX CONCURRENTLY idx_users_id_btree ON users (id);
CREATE INDEX CONCURRENTLY idx_user_roles_composite ON user_roles (user_id, role_id);

-- Analyze primary key usage
SELECT 
    schemaname,
    tablename,
    indexname,
    idx_scan,
    idx_tup_read,
    idx_tup_fetch
FROM pg_stat_user_indexes
WHERE indexname LIKE '%_pkey'
ORDER BY idx_scan DESC;

Foreign Key Constraints¶

Referential Integrity¶

-- Create tables with foreign key relationships
CREATE TABLE categories (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    parent_id INTEGER REFERENCES categories(id) ON DELETE SET NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(200) NOT NULL,
    category_id INTEGER NOT NULL REFERENCES categories(id) ON DELETE RESTRICT,
    price NUMERIC(10,2) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE order_items (
    id SERIAL PRIMARY KEY,
    order_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL REFERENCES products(id) ON DELETE CASCADE,
    quantity INTEGER NOT NULL,
    unit_price NUMERIC(10,2) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- Create foreign key with custom action
CREATE TABLE user_sessions (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    session_token VARCHAR(255) UNIQUE NOT NULL,
    expires_at TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

Foreign Key Performance¶

-- Create indexes for foreign key columns
CREATE INDEX idx_products_category_id ON products (category_id);
CREATE INDEX idx_order_items_order_id ON order_items (order_id);
CREATE INDEX idx_order_items_product_id ON order_items (product_id);
CREATE INDEX idx_user_sessions_user_id ON user_sessions (user_id);

-- Analyze foreign key performance
SELECT 
    tc.table_name,
    kcu.column_name,
    ccu.table_name AS foreign_table_name,
    ccu.column_name AS foreign_column_name,
    tc.constraint_name
FROM information_schema.table_constraints AS tc
JOIN information_schema.key_column_usage AS kcu
    ON tc.constraint_name = kcu.constraint_name
JOIN information_schema.constraint_column_usage AS ccu
    ON ccu.constraint_name = tc.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY'
ORDER BY tc.table_name, kcu.column_name;

Check Constraints¶

Data Validation Constraints¶

-- Create tables with check constraints
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL,
    age INTEGER,
    phone VARCHAR(20),
    status VARCHAR(20) DEFAULT 'active',
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,

    -- Check constraints for data validation
    CONSTRAINT users_username_length CHECK (LENGTH(username) >= 3),
    CONSTRAINT users_email_format CHECK (email ~* '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$'),
    CONSTRAINT users_age_range CHECK (age IS NULL OR (age >= 0 AND age <= 150)),
    CONSTRAINT users_phone_format CHECK (phone IS NULL OR phone ~ '^\+?[1-9]\d{1,14}$'),
    CONSTRAINT users_status_valid CHECK (status IN ('active', 'inactive', 'suspended', 'deleted'))
);

CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(200) NOT NULL,
    price NUMERIC(10,2) NOT NULL,
    stock_quantity INTEGER NOT NULL,
    weight_grams INTEGER,
    dimensions JSONB,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,

    -- Check constraints for business rules
    CONSTRAINT products_price_positive CHECK (price > 0),
    CONSTRAINT products_stock_non_negative CHECK (stock_quantity >= 0),
    CONSTRAINT products_weight_positive CHECK (weight_grams IS NULL OR weight_grams > 0),
    CONSTRAINT products_dimensions_valid CHECK (
        dimensions IS NULL OR 
        (dimensions ? 'length' AND dimensions ? 'width' AND dimensions ? 'height')
    )
);

Complex Check Constraints¶

-- Create complex check constraints
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    order_date DATE NOT NULL,
    ship_date DATE,
    delivery_date DATE,
    total_amount NUMERIC(10,2) NOT NULL,
    status VARCHAR(20) DEFAULT 'pending',
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,

    -- Complex date validation
    CONSTRAINT orders_ship_date_after_order CHECK (
        ship_date IS NULL OR ship_date >= order_date
    ),
    CONSTRAINT orders_delivery_date_after_ship CHECK (
        delivery_date IS NULL OR ship_date IS NULL OR delivery_date >= ship_date
    ),

    -- Status-based validation
    CONSTRAINT orders_status_valid CHECK (
        status IN ('pending', 'processing', 'shipped', 'delivered', 'cancelled', 'returned')
    ),

    -- Business rule validation
    CONSTRAINT orders_amount_positive CHECK (total_amount > 0),
    CONSTRAINT orders_dates_logical CHECK (
        (status = 'pending' AND ship_date IS NULL AND delivery_date IS NULL) OR
        (status = 'processing' AND ship_date IS NULL AND delivery_date IS NULL) OR
        (status = 'shipped' AND ship_date IS NOT NULL AND delivery_date IS NULL) OR
        (status = 'delivered' AND ship_date IS NOT NULL AND delivery_date IS NOT NULL) OR
        (status = 'cancelled' AND ship_date IS NULL AND delivery_date IS NULL) OR
        (status = 'returned' AND ship_date IS NOT NULL AND delivery_date IS NOT NULL)
    )
);

Unique Constraints¶

Unique Constraint Design¶

-- Create unique constraints for data integrity
CREATE TABLE user_emails (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    email VARCHAR(255) NOT NULL,
    is_primary BOOLEAN DEFAULT FALSE,
    verified BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,

    -- Unique constraints
    CONSTRAINT user_emails_email_unique UNIQUE (email),
    CONSTRAINT user_emails_primary_unique UNIQUE (user_id, is_primary) 
        DEFERRABLE INITIALLY DEFERRED
);

-- Partial unique constraints
CREATE TABLE user_sessions (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    session_token VARCHAR(255) NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,

    -- Partial unique constraint
    CONSTRAINT user_sessions_active_unique UNIQUE (user_id, is_active) 
        WHERE is_active = TRUE
);

-- Composite unique constraints
CREATE TABLE user_permissions (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    resource_type VARCHAR(50) NOT NULL,
    resource_id INTEGER NOT NULL,
    permission VARCHAR(50) NOT NULL,
    granted_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,

    -- Composite unique constraint
    CONSTRAINT user_permissions_unique UNIQUE (user_id, resource_type, resource_id, permission)
);

Unique Constraint Performance¶

-- Create indexes for unique constraints
CREATE INDEX CONCURRENTLY idx_user_emails_email ON user_emails (email);
CREATE INDEX CONCURRENTLY idx_user_emails_user_id ON user_emails (user_id);
CREATE INDEX CONCURRENTLY idx_user_sessions_user_id ON user_sessions (user_id);
CREATE INDEX CONCURRENTLY idx_user_permissions_user_id ON user_permissions (user_id);

-- Analyze unique constraint usage
SELECT 
    schemaname,
    tablename,
    indexname,
    idx_scan,
    idx_tup_read,
    idx_tup_fetch
FROM pg_stat_user_indexes
WHERE indexname LIKE '%_unique%'
ORDER BY idx_scan DESC;

Not Null Constraints¶

Not Null Design¶

-- Create tables with appropriate NOT NULL constraints
CREATE TABLE user_profiles (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    first_name VARCHAR(100) NOT NULL,
    last_name VARCHAR(100) NOT NULL,
    email VARCHAR(255) NOT NULL,
    phone VARCHAR(20),
    date_of_birth DATE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- Use NOT NULL with default values
CREATE TABLE system_settings (
    id SERIAL PRIMARY KEY,
    setting_name VARCHAR(100) NOT NULL,
    setting_value TEXT NOT NULL,
    description TEXT,
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- Conditional NOT NULL constraints
CREATE TABLE user_addresses (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    address_type VARCHAR(20) NOT NULL,
    street_address VARCHAR(200) NOT NULL,
    city VARCHAR(100) NOT NULL,
    state VARCHAR(50),
    zip_code VARCHAR(20),
    country VARCHAR(50) NOT NULL,
    is_primary BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,

    -- Conditional NOT NULL based on country
    CONSTRAINT user_addresses_state_required CHECK (
        (country = 'US' AND state IS NOT NULL) OR 
        (country != 'US' AND state IS NULL)
    )
);

Custom Validation Functions¶

Advanced Validation¶

-- Create custom validation functions
CREATE OR REPLACE FUNCTION validate_email(email TEXT)
RETURNS BOOLEAN AS $$
BEGIN
    IF email IS NULL THEN
        RETURN FALSE;
    END IF;

    -- Basic email format validation
    IF NOT (email ~* '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$') THEN
        RETURN FALSE;
    END IF;

    -- Check for consecutive dots
    IF email ~ '\.\.' THEN
        RETURN FALSE;
    END IF;

    -- Check for leading/trailing dots
    IF email ~ '^\.|\.$' THEN
        RETURN FALSE;
    END IF;

    RETURN TRUE;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION validate_phone(phone TEXT)
RETURNS BOOLEAN AS $$
BEGIN
    IF phone IS NULL THEN
        RETURN TRUE; -- NULL is allowed
    END IF;

    -- E.164 format validation
    IF NOT (phone ~ '^\+[1-9]\d{1,14}$') THEN
        RETURN FALSE;
    END IF;

    RETURN TRUE;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION validate_credit_card(card_number TEXT)
RETURNS BOOLEAN AS $$
DECLARE
    i INTEGER;
    sum INTEGER := 0;
    digit INTEGER;
    is_even BOOLEAN := FALSE;
BEGIN
    IF card_number IS NULL THEN
        RETURN FALSE;
    END IF;

    -- Remove non-digits
    card_number := regexp_replace(card_number, '[^0-9]', '', 'g');

    -- Check length
    IF LENGTH(card_number) < 13 OR LENGTH(card_number) > 19 THEN
        RETURN FALSE;
    END IF;

    -- Luhn algorithm
    FOR i IN REVERSE LENGTH(card_number)..1 LOOP
        digit := (ASCII(SUBSTRING(card_number, i, 1)) - ASCII('0'));

        IF is_even THEN
            digit := digit * 2;
            IF digit > 9 THEN
                digit := digit - 9;
            END IF;
        END IF;

        sum := sum + digit;
        is_even := NOT is_even;
    END LOOP;

    RETURN (sum % 10) = 0;
END;
$$ LANGUAGE plpgsql;

-- Use validation functions in constraints
CREATE TABLE user_payments (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    card_number VARCHAR(19) NOT NULL,
    card_holder_name VARCHAR(100) NOT NULL,
    expiry_date DATE NOT NULL,
    cvv VARCHAR(4) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,

    CONSTRAINT user_payments_card_valid CHECK (validate_credit_card(card_number)),
    CONSTRAINT user_payments_expiry_future CHECK (expiry_date > CURRENT_DATE)
);

Constraint Management¶

Dynamic Constraint Management¶

-- Create function to add constraints dynamically
CREATE OR REPLACE FUNCTION add_constraint_if_not_exists(
    table_name TEXT,
    constraint_name TEXT,
    constraint_definition TEXT
)
RETURNS BOOLEAN AS $$
BEGIN
    -- Check if constraint already exists
    IF EXISTS (
        SELECT 1 FROM information_schema.table_constraints 
        WHERE table_name = $1 AND constraint_name = $2
    ) THEN
        RETURN FALSE;
    END IF;

    -- Add constraint
    EXECUTE format('ALTER TABLE %I ADD CONSTRAINT %I %s', 
                   table_name, constraint_name, constraint_definition);

    RETURN TRUE;
END;
$$ LANGUAGE plpgsql;

-- Use function to add constraints
SELECT add_constraint_if_not_exists(
    'users',
    'users_username_format',
    'CHECK (username ~ ''^[a-zA-Z0-9_]{3,50}$'')'
);

-- Create function to drop constraints safely
CREATE OR REPLACE FUNCTION drop_constraint_if_exists(
    table_name TEXT,
    constraint_name TEXT
)
RETURNS BOOLEAN AS $$
BEGIN
    -- Check if constraint exists
    IF NOT EXISTS (
        SELECT 1 FROM information_schema.table_constraints 
        WHERE table_name = $1 AND constraint_name = $2
    ) THEN
        RETURN FALSE;
    END IF;

    -- Drop constraint
    EXECUTE format('ALTER TABLE %I DROP CONSTRAINT %I', table_name, constraint_name);

    RETURN TRUE;
END;
$$ LANGUAGE plpgsql;

Constraint Monitoring¶

# monitoring/constraint_monitor.py
import psycopg2
import json
from datetime import datetime
import logging

class ConstraintMonitor:
    def __init__(self, connection_params):
        self.conn_params = connection_params
        self.setup_logging()

    def setup_logging(self):
        """Setup logging."""
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def get_constraint_info(self):
        """Get information about all constraints."""
        conn = psycopg2.connect(**self.conn_params)

        try:
            with conn.cursor() as cur:
                cur.execute("""
                    SELECT 
                        tc.table_name,
                        tc.constraint_name,
                        tc.constraint_type,
                        kcu.column_name,
                        ccu.table_name AS foreign_table_name,
                        ccu.column_name AS foreign_column_name,
                        rc.delete_rule,
                        rc.update_rule
                    FROM information_schema.table_constraints AS tc
                    LEFT JOIN information_schema.key_column_usage AS kcu
                        ON tc.constraint_name = kcu.constraint_name
                    LEFT JOIN information_schema.constraint_column_usage AS ccu
                        ON ccu.constraint_name = tc.constraint_name
                    LEFT JOIN information_schema.referential_constraints AS rc
                        ON tc.constraint_name = rc.constraint_name
                    WHERE tc.table_schema = 'public'
                    ORDER BY tc.table_name, tc.constraint_type, tc.constraint_name
                """)

                constraints = cur.fetchall()
                return constraints

        except Exception as e:
            self.logger.error(f"Error getting constraint info: {e}")
            return []
        finally:
            conn.close()

    def check_constraint_violations(self, table_name):
        """Check for potential constraint violations."""
        conn = psycopg2.connect(**self.conn_params)

        try:
            with conn.cursor() as cur:
                # Check for duplicate values in unique constraints
                cur.execute("""
                    SELECT 
                        tc.constraint_name,
                        kcu.column_name,
                        COUNT(*) as duplicate_count
                    FROM information_schema.table_constraints tc
                    JOIN information_schema.key_column_usage kcu
                        ON tc.constraint_name = kcu.constraint_name
                    WHERE tc.table_name = %s
                    AND tc.constraint_type = 'UNIQUE'
                    GROUP BY tc.constraint_name, kcu.column_name
                    HAVING COUNT(*) > 1
                """, (table_name,))

                violations = cur.fetchall()
                return violations

        except Exception as e:
            self.logger.error(f"Error checking constraint violations: {e}")
            return []
        finally:
            conn.close()

    def analyze_constraint_performance(self):
        """Analyze constraint performance impact."""
        conn = psycopg2.connect(**self.conn_params)

        try:
            with conn.cursor() as cur:
                # Get constraint usage statistics
                cur.execute("""
                    SELECT 
                        schemaname,
                        tablename,
                        indexname,
                        idx_scan,
                        idx_tup_read,
                        idx_tup_fetch
                    FROM pg_stat_user_indexes
                    WHERE indexname LIKE '%_pkey%' OR indexname LIKE '%_unique%'
                    ORDER BY idx_scan DESC
                """)

                performance_stats = cur.fetchall()
                return performance_stats

        except Exception as e:
            self.logger.error(f"Error analyzing constraint performance: {e}")
            return []
        finally:
            conn.close()

    def generate_constraint_report(self):
        """Generate comprehensive constraint report."""
        constraint_info = self.get_constraint_info()
        performance_stats = self.analyze_constraint_performance()

        report = {
            'report_timestamp': datetime.now().isoformat(),
            'total_constraints': len(constraint_info),
            'constraint_info': constraint_info,
            'performance_stats': performance_stats
        }

        return report

# Usage
if __name__ == "__main__":
    monitor = ConstraintMonitor({
        'host': 'localhost',
        'database': 'production',
        'user': 'monitor_user',
        'password': 'monitor_password'
    })

    report = monitor.generate_constraint_report()
    print(json.dumps(report, indent=2))

Error Handling and Messages¶

Custom Constraint Messages¶

-- Create function for custom constraint error messages
CREATE OR REPLACE FUNCTION handle_constraint_violation()
RETURNS TRIGGER AS $$
BEGIN
    -- Check for specific constraint violations
    IF TG_OP = 'INSERT' THEN
        -- Check for duplicate email
        IF EXISTS (SELECT 1 FROM users WHERE email = NEW.email AND id != NEW.id) THEN
            RAISE EXCEPTION 'Email address % is already in use', NEW.email;
        END IF;

        -- Check for duplicate username
        IF EXISTS (SELECT 1 FROM users WHERE username = NEW.username AND id != NEW.id) THEN
            RAISE EXCEPTION 'Username % is already taken', NEW.username;
        END IF;
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Create trigger for custom error handling
CREATE TRIGGER users_constraint_handler
    BEFORE INSERT OR UPDATE ON users
    FOR EACH ROW EXECUTE FUNCTION handle_constraint_violation();

TL;DR Runbook¶

Quick Start¶

-- 1. Create primary key constraints
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL
);

-- 2. Create foreign key constraints
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    total_amount NUMERIC(10,2) NOT NULL
);

-- 3. Create check constraints
ALTER TABLE users ADD CONSTRAINT users_email_format 
    CHECK (email ~* '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$');

-- 4. Create unique constraints
CREATE UNIQUE INDEX idx_users_username_unique ON users (username);

Essential Patterns¶

# Complete PostgreSQL constraints and validation setup
def setup_postgresql_constraints_validation():
    # 1. Primary key constraints
    # 2. Foreign key constraints
    # 3. Check constraints
    # 4. Unique constraints
    # 5. Not null constraints
    # 6. Custom validation functions
    # 7. Constraint management
    # 8. Error handling and monitoring

    print("PostgreSQL constraints and validation setup complete!")