PostgreSQL Development Environment Best Practices¶

# tests/conftest.py
import pytest
import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT

@pytest.fixture(scope="session")
def test_db():
    """Create test database for the session."""
    conn = psycopg2.connect(
        host="localhost",
        database="postgres",
        user="postgres",
        password="postgres"
    )
    conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)

    with conn.cursor() as cur:
        cur.execute("DROP DATABASE IF EXISTS test_db;")
        cur.execute("CREATE DATABASE test_db;")

    conn.close()

    # Return connection to test database
    test_conn = psycopg2.connect(
        host="localhost",
        database="test_db",
        user="postgres",
        password="postgres"
    )

    yield test_conn
    test_conn.close()

    # Cleanup
    cleanup_conn = psycopg2.connect(
        host="localhost",
        database="postgres",
        user="postgres",
        password="postgres"
    )
    cleanup_conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
    with cleanup_conn.cursor() as cur:
        cur.execute("DROP DATABASE test_db;")
    cleanup_conn.close()

@pytest.fixture
def test_table(test_db):
    """Create test table for each test."""
    with test_db.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS test_table (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100),
                value INTEGER
            );
        """)
        test_db.commit()

    yield test_db

    # Cleanup after each test
    with test_db.cursor() as cur:
        cur.execute("DROP TABLE IF EXISTS test_table;")
        test_db.commit()

# migrations/001_initial_schema.sql
-- Initial schema creation
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE posts (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    title VARCHAR(200) NOT NULL,
    content TEXT,
    published BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create indexes
CREATE INDEX idx_posts_user_id ON posts(user_id);
CREATE INDEX idx_posts_published ON posts(published);
CREATE INDEX idx_posts_created_at ON posts(created_at);

# scripts/seed_data.py
import psycopg2
import uuid
from datetime import datetime, timedelta
import random

def seed_development_data():
    """Seed development database with sample data."""
    conn = psycopg2.connect(
        host="localhost",
        database="development",
        user="dev_user",
        password="dev_password"
    )

    with conn.cursor() as cur:
        # Clear existing data
        cur.execute("TRUNCATE TABLE posts, users RESTART IDENTITY CASCADE;")

        # Insert sample users
        users_data = [
            ('john_doe', 'john@example.com', 'hashed_password_1'),
            ('jane_smith', 'jane@example.com', 'hashed_password_2'),
            ('bob_wilson', 'bob@example.com', 'hashed_password_3'),
            ('alice_brown', 'alice@example.com', 'hashed_password_4'),
            ('charlie_davis', 'charlie@example.com', 'hashed_password_5')
        ]

        for username, email, password_hash in users_data:
            cur.execute("""
                INSERT INTO users (username, email, password_hash)
                VALUES (%s, %s, %s)
            """, (username, email, password_hash))

        # Insert sample posts
        post_titles = [
            'Getting Started with PostgreSQL',
            'Advanced SQL Techniques',
            'Database Design Best Practices',
            'Performance Optimization Tips',
            'Security Considerations',
            'Backup and Recovery Strategies',
            'Monitoring and Maintenance',
            'Scaling PostgreSQL',
            'Replication Setup',
            'Troubleshooting Common Issues'
        ]

        for i in range(50):
            user_id = random.randint(1, 5)
            title = random.choice(post_titles)
            content = f"This is sample content for post {i+1}. " * 10
            published = random.choice([True, False])
            created_at = datetime.now() - timedelta(days=random.randint(0, 30))

            cur.execute("""
                INSERT INTO posts (user_id, title, content, published, created_at)
                VALUES (%s, %s, %s, %s, %s)
            """, (user_id, title, content, published, created_at))

        conn.commit()
        print("Development data seeded successfully!")

    conn.close()

if __name__ == "__main__":
    seed_development_data()

# monitoring/dev_monitor.py
import psycopg2
import time
import json
from datetime import datetime

class PostgreSQLMonitor:
    def __init__(self, connection_params):
        self.conn_params = connection_params
        self.conn = None

    def connect(self):
        """Connect to PostgreSQL."""
        self.conn = psycopg2.connect(**self.conn_params)

    def get_connection_stats(self):
        """Get connection statistics."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    count(*) as total_connections,
                    count(*) FILTER (WHERE state = 'active') as active_connections,
                    count(*) FILTER (WHERE state = 'idle') as idle_connections
                FROM pg_stat_activity;
            """)
            return cur.fetchone()

    def get_database_size(self):
        """Get database size."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT pg_size_pretty(pg_database_size(current_database()));
            """)
            return cur.fetchone()[0]

    def get_table_sizes(self):
        """Get table sizes."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT 
                    schemaname,
                    tablename,
                    pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size
                FROM pg_tables 
                WHERE schemaname = 'public'
                ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC;
            """)
            return cur.fetchall()

    def monitor_loop(self, interval=60):
        """Run monitoring loop."""
        while True:
            try:
                stats = {
                    'timestamp': datetime.now().isoformat(),
                    'connections': self.get_connection_stats(),
                    'database_size': self.get_database_size(),
                    'table_sizes': self.get_table_sizes()
                }

                print(json.dumps(stats, indent=2))
                time.sleep(interval)

            except Exception as e:
                print(f"Monitoring error: {e}")
                time.sleep(interval)

# Usage
if __name__ == "__main__":
    monitor = PostgreSQLMonitor({
        'host': 'localhost',
        'database': 'development',
        'user': 'dev_user',
        'password': 'dev_password'
    })

    monitor.connect()
    monitor.monitor_loop(interval=30)

# Complete PostgreSQL development setup
def setup_postgresql_development():
    # 1. PostgreSQL installation
    # 2. Version management
    # 3. Essential extensions
    # 4. Development configuration
    # 5. Testing environment
    # 6. Development tools
    # 7. Performance monitoring
    # 8. Data management

    print("PostgreSQL development environment setup complete!")