postgresql

name	postgresql
description	Advanced PostgreSQL database management, query optimization, and performance tuning
category	databases

PostgreSQL

What I do

I am a powerful, open-source relational database management system (RDBMS) known for its robustness, extensibility, and standards compliance. I support advanced features including ACID transactions, complex queries, foreign keys, triggers, updatable views, and stored procedures. I excel at handling complex data relationships, supporting JSON/JSONB for document-style data, full-text search, and advanced indexing strategies. I am widely used for mission-critical applications requiring data integrity and complex querying capabilities.

When to use me

• Building enterprise applications requiring ACID compliance and complex transactions
• Applications needing advanced SQL features (window functions, CTEs, recursive queries)
• Systems requiring robust data integrity and referential integrity constraints
• Applications combining relational and document-based data (JSON/JSONB)
• Projects needing strong extensibility with custom types, functions, and extensions
• Data warehousing and analytical workloads with complex aggregations
• Geographic information systems (GIS) with PostGIS extension
• Full-text search applications without external search engines

Core Concepts

1. ACID Transactions: Atomic, Consistent, Isolated, Durable transactions ensure data integrity even during system failures
2. MVCC (Multi-Version Concurrency Control): Allows concurrent reads and writes without locking by maintaining multiple versions of data
3. Index Types: B-tree (default), Hash, GiST (geometric), SP-GiST, GIN (generalized inverted index), and BRIN (block range)
4. JSON/JSONB: Native support for semi-structured data with full query capabilities on JSON documents
5. PostgreSQL Extensions: Modular functionality including PostGIS, pg_cron, pg_partman, and custom extensions
6. Query Optimization: Cost-based optimizer using statistics to choose optimal execution plans
7. Connection Pooling: pgBouncer or built-in connection pooling for managing database connections
8. Replication: Streaming replication, logical replication, and bidirectional replication for high availability
9. Partitioning: Table partitioning for managing large datasets and improving query performance
10. Window Functions: Advanced analytical queries with ROW_NUMBER, RANK, LAG, LEAD functions

Code Examples

Basic Connection and Query Execution

import psycopg2
from psycopg2 import sql
from contextlib import contextmanager

@contextmanager
def get_connection():
    conn = psycopg2.connect(
        host="localhost",
        database="app_db",
        user="app_user",
        password="secure_password",
        port=5432
    )
    try:
        yield conn
    finally:
        conn.close()

def fetch_users_with_orders(limit=100):
    query = """
        SELECT u.id, u.email, u.created_at, COUNT(o.id) as order_count
        FROM users u
        LEFT JOIN orders o ON u.id = o.user_id
        WHERE u.created_at >= %s
        GROUP BY u.id, u.email, u.created_at
        ORDER BY order_count DESC
        LIMIT %s
    """
    with get_connection() as conn:
        with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
            cur.execute(query, ("2024-01-01", limit))
            return cur.fetchall()

def insert_product_with_categories(product_data, category_ids):
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO products (name, description, price, sku)
                VALUES (%s, %s, %s, %s)
                RETURNING id
            """, (product_data["name"], product_data["description"],
                  product_data["price"], product_data["sku"]))
            product_id = cur.fetchone()[]
            
            cur.executemany(, [(product_id, cat_id)  cat_id  category_ids])
            
            conn.commit()
             product_id

Advanced Query with JSONB and Window Functions

import json
from psycopg2.extras import Json

def get_user_analytics():
    query = """
        WITH user_purchases AS (
            SELECT 
                user_id,
                created_at::date as purchase_date,
                SUM(total_amount) as daily_total,
                COUNT(*) as purchase_count
            FROM orders
            WHERE created_at >= CURRENT_DATE - INTERVAL '30 days'
            GROUP BY user_id, created_at::date
        )
        SELECT 
            user_id,
            purchase_date,
            daily_total,
            purchase_count,
            SUM(daily_total) OVER (PARTITION BY user_id ORDER BY purchase_date 
                ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) as rolling_7day_total,
            ROW_NUMBER() OVER (PARTITION BY purchase_date ORDER BY daily_total DESC) as daily_rank
        FROM user_purchases
        ORDER BY user_id, purchase_date
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(query)
            return cur.fetchall()

def search_products_by_attributes(filters):
    where_clauses = []
    params = []
    
    if "category" in filters:
        where_clauses.append("attributes->>'category' = %s")
        params.append(filters["category"])
    
    if "min_price" in filters:
        where_clauses.append("(attributes->>'price')::decimal >= %s")
        params.append(filters["min_price"])
    
    if "tags" in filters:
        where_clauses.append("attributes->'tags' @> %s::jsonb")
        params.append(json.dumps(filters["tags"]))
    
    query = f"""
        SELECT id, name, attributes
        FROM products
        {'WHERE ' + .join(where_clauses)  where_clauses  }
        ORDER BY (attributes->>'popularity')::int DESC
        LIMIT 50
    """
    
     get_connection()  conn:
         conn.cursor()  cur:
            cur.execute(query, params)
             cur.fetchall()

Transaction Management with Savepoints

from psycopg2 import DatabaseError

def process_order_with_inventory(order_data, items):
    with get_connection() as conn:
        with conn.cursor() as cur:
            try:
                cur.execute("BEGIN")
                
                cur.execute("""
                    INSERT INTO orders (user_id, total, status)
                    VALUES (%s, %s, 'pending')
                    RETURNING id
                """, (order_data["user_id"], order_data["total"]))
                order_id = cur.fetchone()[0]
                
                cur.execute("SAVEPOINT before_items")
                
                for item in items:
                    cur.execute("""
                        UPDATE inventory 
                        SET quantity = quantity - %s 
                        WHERE product_id = %s AND quantity >= %s
                    """, (item["quantity"], item["product_id"], item["quantity"]))
                    
                    if cur.rowcount == 0:
                        cur.execute("ROLLBACK TO SAVEPOINT before_items")
                        raise ValueError(f"Insufficient inventory for product {item['product_id']}")
                    
                    cur.execute("""
                        INSERT INTO order_items (order_id, product_id, quantity, price)
                        VALUES (%s, %s, %s, %s)
                    """, (order_id, item["product_id"], item["quantity"], item["price"]))
                
                cur.execute("COMMIT")
                return order_id
                
             (DatabaseError, ValueError)  e:
                cur.execute()
                 e

Using COPY for Bulk Operations

import io
from psycopg2.extras import execute_values

def bulk_insert_products(products):
    with get_connection() as conn:
        with conn.cursor() as cur:
            execute_values(cur, """
                INSERT INTO products (name, sku, price, category, attributes, created_at)
                VALUES %s
                ON CONFLICT (sku) DO UPDATE SET
                    name = EXCLUDED.name,
                    price = EXCLUDED.price,
                    attributes = EXCLUDED.attributes,
                    updated_at = CURRENT_TIMESTAMP
            """, products)
            conn.commit()

def export_orders_to_csv(start_date, end_date):
    query = """
        COPY (
            SELECT o.id, o.created_at, u.email, p.name, oi.quantity, oi.price
            FROM orders o
            JOIN users u ON o.user_id = u.id
            JOIN order_items oi ON o.id = oi.order_id
            JOIN products p ON oi.product_id = p.id
            WHERE o.created_at BETWEEN %s AND %s
            ORDER BY o.created_at
        ) TO STDOUT WITH CSV HEADER
    """
    with get_connection() as conn:
        with conn.cursor() as cur:
            buffer = io.StringIO()
            cur.copy_expert(query, buffer, size=8192)
            return buffer.getvalue()

def import_products_from_csv(csv_file):
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TEMP TABLE temp_products (
                    name TEXT, sku TEXT, price DECIMAL, category TEXT
                ) ON COMMIT DROP
            """)
            buffer = io.StringIO(csv_file)
            cur.copy_from(buffer, "temp_products", columns=("name", , , ), sep=)
            
            cur.execute()
            conn.commit()

Using Asynchronous Connections

import asyncio
import asyncpg

async def fetch_user_stats(user_id):
    conn = await asyncpg.connect(
        host="localhost",
        database="app_db",
        user="app_user",
        password="secure_password"
    )
    try:
        async with conn.transaction():
            user = await conn.fetchrow("""
                SELECT id, email, created_at FROM users WHERE id = $1
            """, user_id)
            
            orders = await conn.fetch("""
                SELECT COUNT(*) as total_orders, SUM(total) as total_spent
                FROM orders WHERE user_id = $1
            """, user_id)
            
            return {"user": dict(user), "orders": dict(orders[0])}
    finally:
        await conn.close()

async def bulk_insert_events(events):
    conn = await asyncpg.connect(
        host="localhost",
        database="app_db",
        user="app_user",
        password="secure_password"
    )
    try:
        await conn.executemany("""
            INSERT INTO events (user_id, event_type, metadata, created_at)
            VALUES ($1, $2, $3, $4)
        """, events)
    finally:
        await conn.close()

async def ():
    conn =  asyncpg.connect(
        host=,
        database=,
        user=,
        password=
    )
    :
        queries = [
            (, ),
            (, ),
            (, ),
            (, )
        ]
        
        results =  asyncio.gather(
            *[conn.fetchval(query)  query, _  queries]
        )
        
         (([name  _, name  queries], results))
    :
         conn.close()

Best Practices

1. Use Connection Pooling: Implement connection pooling with pgBouncer to handle connection overhead and prevent connection exhaustion under load
2. Optimize Indexes: Create appropriate indexes based on query patterns; use EXPLAIN ANALYZE to identify missing indexes
3. Partition Large Tables: Use table partitioning for tables exceeding 10GB or with clear partition keys (time-based, range-based)
4. Implement Proper Backups: Use pg_basebackup for physical backups and pg_dump for logical backups; test restoration regularly
5. Configure Appropriate Checkpoint Intervals: Tune checkpoint_completion_target and checkpoint_segments for write-heavy workloads
6. Use Prepared Statements: For frequently executed queries, use prepared statements to reduce planning overhead
7. Monitor Query Performance: Track slow queries using pg_stat_statements and set appropriate log_min_duration_statement
8. Implement Proper Vacuuming: Configure autovacuum appropriately and run manual VACUUM ANALYZE when needed
9. Use Proper Data Types: Choose appropriate data types (INT vs BIGINT, VARCHAR limits, JSONB vs TEXT) for storage efficiency
10. Secure Your Instance: Use SSL/TLS for connections, implement proper authentication (md5, scram-sha-256), and restrict network access