Postgres Patterns

Advanced PostgreSQL design patterns for automated data modeling and high-efficiency query integration

Category: development Source: affaan-m/everything-claude-code

Postgres Patterns is an AI skill that provides database design patterns and query optimization strategies for PostgreSQL applications. It covers schema design, indexing strategies, query performance tuning, partitioning, connection management, and migration workflows that enable developers to build efficient and scalable database layers.

What Is This?

Overview

Postgres Patterns provides structured approaches to designing and optimizing PostgreSQL databases. It handles designing normalized schemas with appropriate constraints and relationships, creating indexes that match query access patterns including composite and partial indexes, writing efficient queries using CTEs, window functions, and lateral joins, implementing table partitioning for large datasets to improve query performance, managing connection pools to prevent exhaustion under load, and planning schema migrations that minimize downtime during deployment.

Who Should Use This

This skill serves backend developers designing database schemas, database administrators optimizing query performance, platform engineers managing PostgreSQL infrastructure, and teams migrating from other databases to PostgreSQL.

Why Use It?

Problems It Solves

Missing indexes cause full table scans that degrade query performance as data grows. Poorly designed schemas lead to complex joins and denormalization problems. Without connection pooling, concurrent requests exhaust available connections and cause timeouts. Schema migrations without planning cause downtime during deployment.

Core Highlights

Index strategy matching aligns index types with actual query patterns for optimal lookups. Query optimization reduces execution time through plan analysis and rewriting. Partitioning distributes large tables for improved scan performance and maintenance. Migration planning enables schema changes with minimal application downtime.

How to Use It?

Basic Usage

-- Composite index for common query pattern
CREATE INDEX idx_orders_user_date
ON orders (user_id, created_at DESC);

-- Partial index for active records only
CREATE INDEX idx_active_subscriptions
ON subscriptions (user_id)
WHERE status = 'active';

-- CTE for readable complex queries
WITH monthly_totals AS (
    SELECT
        user_id,
        DATE_TRUNC('month', created_at) AS month,
        SUM(amount) AS total
    FROM orders
    WHERE created_at >= NOW() - INTERVAL '12 months'
    GROUP BY user_id, DATE_TRUNC('month', created_at)
)
SELECT
    user_id,
    AVG(total) AS avg_monthly_spend,
    MAX(total) AS peak_month
FROM monthly_totals
GROUP BY user_id
HAVING AVG(total) > 100
ORDER BY avg_monthly_spend DESC;

-- Window function for running totals
SELECT
    id, amount,
    SUM(amount) OVER (
        PARTITION BY user_id
        ORDER BY created_at
    ) AS running_total
FROM orders;

Real-World Examples

import psycopg2
from psycopg2 import pool
from contextlib import contextmanager

class PgConnectionPool:
    def __init__(self, dsn, min_conn=2, max_conn=10):
        self.pool = pool.ThreadedConnectionPool(
            min_conn, max_conn, dsn
        )

    @contextmanager
    def connection(self):
        conn = self.pool.getconn()
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            self.pool.putconn(conn)

    @contextmanager
    def cursor(self):
        with self.connection() as conn:
            cur = conn.cursor()
            try:
                yield cur
            finally:
                cur.close()

class QueryAnalyzer:
    def __init__(self, db):
        self.db = db

    def explain(self, query, params=None):
        with self.db.cursor() as cur:
            cur.execute(
                f"EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {query}",
                params
            )
            plan = cur.fetchone()[0]
            return plan

    def find_slow_queries(self):
        with self.db.cursor() as cur:
            cur.execute("""
                SELECT query, calls, mean_exec_time,
                       total_exec_time
                FROM pg_stat_statements
                ORDER BY mean_exec_time DESC
                LIMIT 10
            """)
            return cur.fetchall()

db = PgConnectionPool("postgresql://localhost/mydb")
analyzer = QueryAnalyzer(db)
plan = analyzer.explain(
    "SELECT * FROM orders WHERE user_id = %s", ("123",)
)
print(plan)

Advanced Tips

Use EXPLAIN ANALYZE to verify that queries use intended indexes rather than relying on assumptions. Apply partial indexes for columns with skewed value distributions where most queries target a subset of rows. Monitor pg_stat_statements for slow query identification and optimization targeting.

When to Use It?

Use Cases

Use Postgres Patterns when designing schemas for new applications with anticipated growth, when optimizing slow queries identified through monitoring, when implementing table partitioning for large time-series datasets, or when planning zero-downtime schema migrations.

Important Notes

Requirements

PostgreSQL 14 or later for current feature support. Client library with connection pooling. Monitoring access to pg_stat_statements for query analysis.

Usage Recommendations

Do: analyze query plans before and after adding indexes to verify performance improvement. Use connection pooling in every application to prevent connection exhaustion. Test schema migrations on staging with production-scale data before deploying.

Don't: add indexes on every column without analyzing actual query patterns. Run ALTER TABLE operations on large tables without considering lock implications. Skip connection pool configuration, which causes connection exhaustion under concurrent load.

Limitations

Index maintenance adds write overhead proportional to the number of indexes per table. Query plan optimization depends on accurate table statistics from regular ANALYZE runs. Partitioning adds complexity to schema design and query routing.