# Postgresql

> Advanced PostgreSQL database management, query optimization, and performance tuning

- Skill: `ffsshhttiikk/postgresql-2` (Agent Skill)
- Install (CLI): `npx skillmds@latest add ffsshhttiikk/postgresql-2`
- Raw SKILL.md: https://api.skillmd.com/api/skills/ffsshhttiikk/postgresql-2/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: ffsshhttiikk (https://skillmd.com/u/ffsshhttiikk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/ffsshhttiikk/postgresql-2

---

# 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

```python
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()[0]
            
            cur.executemany("""
                INSERT INTO product_categories (product_id, category_id)
                VALUES (%s, %s)
            """, [(product_id, cat_id) for cat_id in category_ids])
            
            conn.commit()
            return product_id
```

### Advanced Query with JSONB and Window Functions

```python
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 ' + ' AND '.join(where_clauses) if where_clauses else ''}
        ORDER BY (attributes->>'popularity')::int DESC
        LIMIT 50
    """
    
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(query, params)
            return cur.fetchall()
```

### Transaction Management with Savepoints

```python
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
                
            except (DatabaseError, ValueError) as e:
                cur.execute("ROLLBACK")
                raise e
```

### Using COPY for Bulk Operations

```python
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", "sku", "price", "category"), sep=",")
            
            cur.execute("""
                INSERT INTO products (name, sku, price, category)
                SELECT name, sku, price, category FROM temp_products
                ON CONFLICT (sku) DO NOTHING
            """)
            conn.commit()
```

### Using Asynchronous Connections

```python
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 get_dashboard_metrics():
    conn = await asyncpg.connect(
        host="localhost",
        database="app_db",
        user="app_user",
        password="secure_password"
    )
    try:
        queries = [
            ("SELECT COUNT(*) FROM users", "total_users"),
            ("SELECT COUNT(*) FROM orders WHERE created_at > CURRENT_DATE", "today_orders"),
            ("SELECT SUM(total) FROM orders WHERE created_at > CURRENT_DATE", "today_revenue"),
            ("SELECT COUNT(*) FROM products WHERE stock < 10", "low_stock_products")
        ]
        
        results = await asyncio.gather(
            *[conn.fetchval(query) for query, _ in queries]
        )
        
        return dict(zip([name for _, name in queries], results))
    finally:
        await 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