Database Connection Pooling: Patterns, Implementation, and Best Practices

connection_costs = {
    "tcp_handshake": "3-10ms",
    "tls_handshake": "10-50ms",
    "postgresql_startup": "5-20ms",
    "mysql_handshake": "2-10ms",
    "application_query": "1-5ms"
}

# Without pooling (per request):
cost_without_pooling = sum(connection_costs.values())  # ~20-95ms

# With pooling (after first request):
cost_with_pooling = connection_costs["application_query"]  # 1-5ms

# Improvement: 10-50x faster!

pool_configuration = {
    "min_size": {
        "description": "Minimum connections to keep open",
        "typical": "5-10 connections",
        "benefit": "Handles baseline load"
    },
    
    "max_size": {
        "description": "Maximum connections allowed",
        "typical": "20-50 connections",
        "warning": "Don't exceed database limit"
    },
    
    "max_overflow": {
        "description": "Extra connections when pool exhausted",
        "typical": "10-20",
        "use": "Burst traffic handling"
    },
    
    "pool_timeout": {
        "description": "Wait time for available connection",
        "typical": "30 seconds",
        "use": "Fail fast instead of infinite wait"
    },
    
    "pool_recycle": {
        "description": "Recycle connections after N seconds",
        "typical": "3600 seconds (1 hour)",
        "use": "Prevent stale connections"
    },
    
    "pool_pre_ping": {
        "description": "Test connection before use",
        "typical": "True",
        "use": "Detect broken connections"
    }
}

# Pool states

pool_states = {
    "checkout": "Get connection from pool",
    "checkin": "Return connection to pool",
    "overflow": "Create temporary extra connections",
    "timeout": "Wait for available connection",
    "eviction": "Remove idle connections"
}

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool, NullPool

# Basic connection pool
engine = create_engine(
    "postgresql://user:pass@localhost/mydb",
    poolclass=QueuePool,
    pool_size=10,              # Min connections
    max_overflow=20,           # Max overflow
    pool_timeout=30,           # Wait timeout
    pool_recycle=3600,        # Recycle after 1 hour
    pool_pre_ping=True        # Test connections
)

# Using the connection
with engine.connect() as conn:
    result = conn.execute("SELECT * FROM users")
    # Connection automatically returned to pool

# For high-performance scenarios
fast_engine = create_engine(
    "postgresql://user:pass@localhost/mydb",
    poolclass=NullPool,      # No pooling (for async/forked)
    pool_pre_ping=True
)

# Connection per request (for serverless)
serverless_engine = create_engine(
    "postgresql://user:pass@localhost/mydb",
    pool_size=1,
    max_overflow=0
)

from sqlalchemy.orm import sessionmaker

# Create session factory
Session = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False
)

# Use session
def get_user(user_id):
    session = Session()
    try:
        user = session.query(User).filter_by(id=user_id).first()
        return user
    finally:
        session.close()  # Returns connection to pool

# Context manager (recommended)
from contextlib import contextmanager

@contextmanager
def get_session():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

import asyncio
import asyncpg
from asyncpg import Pool

async def create_pool():
    """Create async connection pool"""
    
    pool = await Pool(
        host='localhost',
        port=5432,
        user='user',
        password='password',
        database='mydb',
        min_size=10,
        max_size=20,
        command_timeout=60,
        max_queries=50000,
        max_inactive_connection_lifetime=300
    )
    
    return pool

async def query_example(pool):
    """Using async pool"""
    
    # Get connection from pool
    async with pool.acquire() as conn:
        # Run query
        result = await conn.fetch('SELECT * FROM users')
        
        # Or use prepared statement
        stmt = await conn.prepare('SELECT * FROM users WHERE id = $1')
        user = await stmt.fetchval(1)
    
    # Connection automatically returned to pool

async def main():
    pool = await create_pool()
    
    try:
        await query_example(pool)
    finally:
        await pool.close()

# PostgreSQL connection pooling options

pooling_options:
  - name: "Application-level (SQLAlchemy)"
    description: "Pool in application code"
    pros: "Simple, no extra infrastructure"
    cons: "Each app has its own pool"
    
  - name: "PgBouncer"
    description: "Middleman connection pooler"
    pros: "Shared pool, lightweight"
    cons: "Extra component to manage"
    
  - name: "PgPool-II"
    description: "More than pooling (replication, etc)"
    pros: "Feature-rich"
    cons: "Complex, resource-heavy"

# pgbouncer.ini

[databases]
mydb = host=localhost dbname=mydb

[pgbouncer]
listen_addr = 127.0.0.1
listen_port = 6432
auth_type = md5
auth_file = userlist.txt

pool_mode = transaction  # or session
max_client_conn = 1000
default_pool_size = 20
min_pool_size = 5
reserve_pool_size = 5
reserve_pool_timeout = 5
max_db_connections = 50
max_user_connections = 50

# Using PgBouncer
engine = create_engine(
    "postgresql://user:pass@localhost:6432/mydb"
    # Same code, just different port!
)

# MySQL with PyMySQL
import pymysql
from DBUtils.PooledDB import PooledDB

pool = PooledDB(
    creator=pymysql,
    maxconnections=20,
    mincached=5,
    maxcached=10,
    blocking=True,
    maxusage=None,
    setsession=[],
    ping=1,
    host='localhost',
    port=3306,
    user='user',
    password='password',
    database='mydb',
    charset='utf8mb4'
)

# Use pool
db = pool.connection()
cursor = db.cursor()
cursor.execute("SELECT * FROM users")
result = cursor.fetchall()

# Return to pool
cursor.close()
db.close()

# Serverless connection pooling challenges

serverless_issues = {
    "problem": "Each invocation creates new lambda/container",
    "impact": "Connection exhaustion, cold starts"
}

# Solutions:
solutions = {
    "outside_handler": "Create pool at module level (persists)"
}

# Example: Lambda with connection reuse
import os
import sqlalchemy

# Module level (reused across invocations)
engine = None

def get_engine():
    global engine
    if engine is None:
        engine = create_engine(
            os.environ["DATABASE_URL"],
            pool_size=1,
            max_overflow=0
        )
    return engine

def handler(event, context):
    engine = get_engine()
    with engine.connect() as conn:
        # query
        pass

# Microservice connection strategy

microservice_strategy = {
    "each_service": "Own connection pool",
    
    "shared_pooler": "PgBouncer as sidecar",
    
    "connection_limit": "Calculate per service",
    
    "calculation": "total_db_connections / num_services"
}

# Example: Kubernetes with PgBouncer sidecar
deployment = """
apiVersion: apps/v1
kind: Deployment
spec:
  containers:
  - name: app
    image: myapp
    env:
    - name: DATABASE_URL
      value: "postgresql://localhost:6432/mydb"
  
  - name: pgbouncer
    image: pgbouncer/pgbouncer
    env:
    - name: DATABASE_URL
      value: "postgresql://db:5432/mydb"
    ports:
    - containerPort: 6432
"""

# SQLAlchemy pool monitoring

from sqlalchemy import event
from sqlalchemy.pool import QueuePool

# Monitor checkout times
@event.listens_for(QueuePool, "checkout")
def receive_checkout(dbapi_connection, connection_record, connection_proxy):
    print(f"Connection checked out: {connection_record}")

# Monitor checkins  
@event.listens_for(QueuePool, "checkin")
def receive_checkin(dbapi_connection, connection_record):
    print(f"Connection returned: {connection_record}")

# Get pool status
def get_pool_status(engine):
    pool = engine.pool
    
    return {
        "size": pool.size(),
        "checked_in": pool.checkedin(),
        "checked_out": pool.checkedout(),
        "overflow": pool.overflow(),
        "invalid": pool.invalidatedcount() if hasattr(pool, 'invalidatedcount') else 0
    }

# Connection pool tuning

tuning_tips = {
    "right_size": {
        "description": "Calculate based on actual usage",
        "formula": "concurrent_requests * avg_query_time / 1000"
    },
    
    "monitor": {
        "description": "Track pool saturation",
        "metrics": [
            "wait_time (should be near 0)",
            "timeout_count",
            "connection_age"
        ]
    },
    
    "database_limits": {
        "description": "Know your database limits",
        "postgresql": "max_connections default 100",
        "mysql": "max_connections default 151"
    },
    
    "pre_ping": {
        "description": "Always enable for reliability",
        "benefit": "Detect broken connections"
    }
}

# Do:
- Always use connection pooling in production
- Set appropriate pool size based on load
- Enable pool_pre_ping=True
- Set reasonable timeouts
- Monitor pool metrics
- Handle pool exhaustion gracefully

# Don't:
- Set pool_size too high (exhaust DB limits)
- Use pool for long-running operations
- Forget to return connections
- Ignore pool-related errors
- Use default settings without tuning

# Handle pool exhaustion

from sqlalchemy.exc import TimeoutError
import time
import logging

def robust_query(engine, max_retries=3):
    """Query with retry on pool exhaustion"""
    
    for attempt in range(max_retries):
        try:
            with engine.connect() as conn:
                return conn.execute("SELECT 1").fetchall()
                
        except TimeoutError:
            logging.warning(f"Pool timeout, attempt {attempt + 1}")
            time.sleep(1)  # Wait before retry
            
        except Exception as e:
            logging.error(f"Query error: {e}")
            raise
    
    raise Exception("Max retries exceeded")