System Design (HLD)Connection Pooling

Connection Pooling: Managing Database Connections at Scale

LevelIntermediate

Duration75 mins

TopicConnection Pooling

5 / 5

Application-side Pooling

Mastering Connection Management in Your Application

External poolers like PgBouncer and ProxySQL handle connection aggregation across application instances, but application-side pooling remains essential. Even when using external poolers, your application needs to manage its local pool effectively—handling connection checkout, error recovery, and resource cleanup.

This page focuses on advanced application-side pooling patterns: how to architect your code for robust connection management, handle edge cases gracefully, and integrate pooling with modern application patterns like dependency injection, middleware, and graceful shutdown.

What You Will Learn

By the end of this page, you will understand advanced application-side pooling patterns, connection lifecycle management, error handling strategies, integration with frameworks and dependency injection, testing patterns, and troubleshooting common pool issues.

The Application Pool Architecture

A well-designed application pool isn't just a collection of connections—it's a managed resource with clear lifecycle, health monitoring, and failure recovery. Understanding the internal architecture helps you configure and troubleshoot effectively.

Core pool components:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
┌─────────────────────────────────────────────────────────────────┐
│                    Application Process                          │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                  Connection Pool                        │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Idle Connection Queue              │   │   │
│  │  │  [Conn1] [Conn2] [Conn3] [Conn4] [Conn5]       │   │   │
│  │  │  Ready for immediate checkout                   │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Active Connections                 │   │   │
│  │  │  [Conn6 → Request A] [Conn7 → Request B]       │   │   │
│  │  │  Currently executing queries                    │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Wait Queue (if pool exhausted)      │   │   │
│  │  │  [Request C waiting] [Request D waiting]        │   │   │
│  │  │  FIFO queue for fairness                        │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Background Services                │   │   │
│  │  │  • Connection validator (periodic health check)  │   │   │
│  │  │  • Connection killer (max lifetime enforcement) │   │   │
│  │  │  • Pool scaler (min/max adjustment)             │   │   │
│  │  │  • Metrics collector                            │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Connection lifecycle states:

Connection States in Application Pool
State	Description	Transitions
Creating	Connection being established to database	→ Idle (success) or Destroyed (failure)
Idle	In pool, ready for checkout	→ Active (checkout) or Validating (health check)
Validating	Being tested for health	→ Idle (valid) or Destroyed (invalid)
Active	Checked out, executing queries	→ Idle (return) or Destroyed (error/leak)
Retiring	Past max lifetime, waiting to close	→ Destroyed (when returned or timeout)
Destroyed	Closed, removed from pool	New connection Created to replace

The Validation Tradeoff

Connection validation (running SELECT 1 before returning to caller) adds latency but catches broken connections. Test-on-borrow is safest but slowest. Background validation is efficient but may return stale connections. Test-on-return catches issues early. Most production systems use background validation plus test-on-borrow after idle timeout.

Connection Checkout Patterns

How you acquire connections from the pool significantly impacts reliability and performance. There are several patterns to consider.

Pattern 1: Simple Checkout/Return

The basic pattern where the calling code explicitly manages connection lifecycle.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Explicit checkout/return - requires discipline
class UserRepository:
    def __init__(self, pool):
        self.pool = pool
    
    def get_user(self, user_id: int) -> User:
        conn = self.pool.getconn()      # Checkout
        try:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM users WHERE id = %s", 
                (user_id,)
            )
            row = cursor.fetchone()
            return User.from_row(row) if row else None
        finally:
            self.pool.putconn(conn)     # Always return!

Pattern 2: Context Manager Pattern

Automatic resource management using language-specific constructs.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from contextlib import contextmanager
 
class DatabasePool:
    def __init__(self, connection_string):
        self._pool = psycopg_pool.ConnectionPool(connection_string)
    
    @contextmanager
    def connection(self):
        """Automatic checkout/return with context manager."""
        conn = self._pool.getconn()
        try:
            yield conn
            conn.commit()       # Auto-commit on success
        except Exception:
            conn.rollback()     # Auto-rollback on error
            raise
        finally:
            self._pool.putconn(conn)   # Always return
 
# Usage - impossible to leak connections
def get_user(db: DatabasePool, user_id: int) -> User:
    with db.connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        row = cursor.fetchone()
        return User.from_row(row) if row else None
    # Connection automatically returned here

Pattern 3: Decorator Pattern

Wrap functions with automatic connection injection.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from functools import wraps
 
# Global pool instance (initialized at startup)
db_pool: DatabasePool = None
 
def with_connection(func):
    """Inject connection as first argument."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        with db_pool.connection() as conn:
            return func(conn, *args, **kwargs)
    return wrapper
 
@with_connection
def get_user(conn, user_id: int) -> User:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    row = cursor.fetchone()
    return User.from_row(row) if row else None
 
# Caller doesn't manage connection at all
user = get_user(123)

Pattern 4: Request-Scoped Pattern

Connection shared across a single request, common in web frameworks.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from fastapi import Depends, FastAPI
from contextlib import asynccontextmanager
 
app = FastAPI()
pool: asyncpg.Pool = None
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool
    pool = await asyncpg.create_pool(
        "postgresql://user:pass@db:5432/myapp",
        min_size=5,
        max_size=20
    )
    yield
    await pool.close()
 
app = FastAPI(lifespan=lifespan)
 
async def get_connection():
    """Dependency injection for request-scoped connection."""
    async with pool.acquire() as conn:
        yield conn
 
@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    conn = Depends(get_connection)  # Injected, auto-returned
):
    row = await conn.fetchrow(
        "SELECT * FROM users WHERE id = $1", 
        user_id
    )
    return dict(row) if row else None

One Connection Per Request

For most web applications, the request-scoped pattern is ideal: checkout a connection at request start, use it for all operations, return it at request end. This ensures transactional consistency across operations and simplifies connection management.

Error Handling and Recovery

Database connections fail. Networks partition. Servers restart. Your application must handle these failures gracefully without leaking connections or corrupting state.

Classifying connection errors:

Connection Error Types and Handling
Error Type	Examples	Recovery Strategy
Transient	Network glitch, server briefly unavailable	Retry with backoff, discard connection
Connection broken	Server restart, firewall timeout, network change	Discard connection, create new one
Pool exhausted	All connections in use, wait queue full	Fail fast with clear error, don't block
Authentication	Invalid credentials, permission revoked	Fail immediately, alert operators
Query/Application	Syntax error, constraint violation	Return connection to pool, propagate error

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import time
import random
from contextlib import contextmanager
 
class PoolExhaustedError(Exception):
    """Raised when pool cannot provide connection in time."""
    pass
 
class DatabasePool:
    def __init__(self, pool, max_retries=3, base_delay=0.1):
        self._pool = pool
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    @contextmanager
    def connection(self, retry=True):
        """Get connection with error handling and optional retry."""
        conn = None
        last_error = None
        
        for attempt in range(self.max_retries if retry else 1):
            try:
                # Attempt to get connection with timeout
                conn = self._pool.getconn(timeout=5.0)
                
                # Validate connection before use
                self._validate_connection(conn)
                
                try:
                    yield conn
                    conn.commit()  # Commit on success
                    return
                except Exception as e:
                    self._handle_query_error(conn, e)
                    raise
                    
            except psycopg.PoolTimeout:
                raise PoolExhaustedError(
                    "Could not acquire connection from pool within timeout"
                )
            except psycopg.OperationalError as e:
                last_error = e
                if conn:
                    self._discard_connection(conn)
                    conn = None
                
                if attempt < self.max_retries - 1:
                    delay = self.base_delay * (2 ** attempt) + random.uniform(0, 0.1)
                    time.sleep(delay)
                    continue
                raise
            finally:
                if conn:
                    try:
                        self._pool.putconn(conn)
                    except Exception:
                        pass  # Connection already broken/closed
    
    def _validate_connection(self, conn):
        """Quick health check before use."""
        try:
            cursor = conn.cursor()
            cursor.execute("SELECT 1")
            cursor.fetchone()
        except Exception:
            raise psycopg.OperationalError("Connection validation failed")
    
    def _handle_query_error(self, conn, error):
        """Handle errors during query execution."""
        try:
            conn.rollback()
        except Exception:
            # Connection broken during rollback - mark for discard
            self._discard_connection(conn)
    
    def _discard_connection(self, conn):
        """Mark connection as broken so pool discards it."""
        # Implementation depends on pool library
        # Most pools detect broken connections on return
        pass

Never Swallow Connection Errors

Silently catching connection errors without proper cleanup is a guaranteed way to exhaust your pool. Every exception path must ensure the connection is either returned to the pool or explicitly marked as broken. Use context managers and try/finally religiously.

Transaction Management

Transactions and connection pooling interact in subtle ways. Managing transactions correctly is essential for data integrity and pool efficiency.

The transaction-connection binding:

A database connection can only have one active transaction at a time. When you start a transaction, that connection is "locked" to your operation until commit or rollback. This has implications for pooling:

Transaction-Pool Implications

•Long transactions hold connections — A 30-second transaction keeps that connection unavailable for 30 seconds, reducing effective pool size.
•Transaction leaks are pool leaks — Starting a transaction and not completing it (via exception without rollback) leaves the connection unusable.
•Autocommit for reads — Read-only queries without explicit transactions should use autocommit mode for maximum pool efficiency.
•Transaction scope should be minimal — Do non-database work outside transactions to minimize connection hold time.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
from contextlib import contextmanager
 
class TransactionManager:
    def __init__(self, pool):
        self.pool = pool
    
    @contextmanager
    def transaction(self, isolation_level="READ COMMITTED"):
        """Explicit transaction with guaranteed cleanup."""
        conn = self.pool.getconn()
        conn.set_session(isolation_level=isolation_level, autocommit=False)
        
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.set_session(autocommit=True)  # Reset to autocommit
            self.pool.putconn(conn)
    
    @contextmanager
    def read_only(self):
        """Optimized for read-only operations."""
        conn = self.pool.getconn()
        conn.set_session(readonly=True, autocommit=True)
        
        try:
            yield conn
        finally:
            conn.set_session(readonly=False)
            self.pool.putconn(conn)
 
# Usage
def transfer_money(tx_manager, from_id, to_id, amount):
    with tx_manager.transaction(isolation_level="SERIALIZABLE") as conn:
        cursor = conn.cursor()
        
        # Debit source account
        cursor.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_id)
        )
        
        # Credit destination account
        cursor.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_id)
        )
        
        # Verify source had sufficient balance
        cursor.execute(
            "SELECT balance FROM accounts WHERE id = %s",
            (from_id,)
        )
        if cursor.fetchone()[0] < 0:
            raise ValueError("Insufficient funds")
        
        # Transaction commits automatically via context manager
 
def get_account_summary(tx_manager, account_id):
    with tx_manager.read_only() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT balance, last_transaction FROM accounts WHERE id = %s",
            (account_id,)
        )
        return cursor.fetchone()

Keep Transactions Short

A golden rule: minimize transaction duration. Fetch data, do computation, validate business rules, prepare all data—then open a transaction, execute operations, commit. Don't do network calls, file I/O, or complex computation inside transactions.

Dependency Injection Integration

Modern applications use dependency injection for clean architecture and testability. Integrating connection pools with DI requires careful consideration of scopes and lifecycle.

Java (Spring Framework):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@Configuration
public class DatabaseConfig {
    
    @Bean
    @ConfigurationProperties("spring.datasource.hikari")
    public HikariConfig hikariConfig() {
        return new HikariConfig();
    }
    
    @Bean(destroyMethod = "close")
    public DataSource dataSource(HikariConfig config) {
        // Pool created as singleton, shared across entire application
        return new HikariDataSource(config);
    }
    
    @Bean
    public JdbcTemplate jdbcTemplate(DataSource dataSource) {
        return new JdbcTemplate(dataSource);
    }
    
    @Bean
    public TransactionManager transactionManager(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }
}
 
// Repository using injected JdbcTemplate
@Repository
public class UserRepository {
    private final JdbcTemplate jdbc;
    
    public UserRepository(JdbcTemplate jdbc) {
        this.jdbc = jdbc;  // Constructor injection
    }
    
    public User findById(Long id) {
        // JdbcTemplate handles connection checkout/return internally
        return jdbc.queryForObject(
            "SELECT * FROM users WHERE id = ?",
            new UserRowMapper(),
            id
        );
    }
}

Node.js (NestJS):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import { Module, Global, OnApplicationShutdown } from '@nestjs/common';
import { Pool } from 'pg';
 
const DATABASE_POOL = 'DATABASE_POOL';
 
@Global()
@Module({
  providers: [
    {
      provide: DATABASE_POOL,
      useFactory: async () => {
        const pool = new Pool({
          host: process.env.DB_HOST,
          port: parseInt(process.env.DB_PORT),
          database: process.env.DB_NAME,
          user: process.env.DB_USER,
          password: process.env.DB_PASSWORD,
          max: 20,
          idleTimeoutMillis: 30000,
          connectionTimeoutMillis: 10000,
        });
        
        // Validate pool is working
        const client = await pool.connect();
        client.release();
        
        return pool;
      },
    },
  ],
  exports: [DATABASE_POOL],
})
export class DatabaseModule implements OnApplicationShutdown {
  constructor(
    @Inject(DATABASE_POOL) private readonly pool: Pool
  ) {}
  
  async onApplicationShutdown(signal?: string) {
    await this.pool.end();
    console.log('Database pool closed');
  }
}
 
// Service using injected pool
@Injectable()
export class UserService {
  constructor(
    @Inject(DATABASE_POOL) private readonly pool: Pool
  ) {}
  
  async findById(id: number): Promise<User | null> {
    const result = await this.pool.query(
      'SELECT * FROM users WHERE id = $1',
      [id]
    );
    return result.rows[0] || null;
  }
}

Pool as Singleton

In DI containers, the database pool should always be a singleton with application lifecycle. Creating new pools per request or per service instance defeats the purpose of pooling. The pool is infrastructure—share it across the entire application.

Graceful Shutdown

When your application stops—for deployment, scaling, or failure—it must close database connections properly. Improper shutdown leaves connections orphaned on the database server.

The graceful shutdown sequence:

Shutdown Steps

•Stop accepting new requests — Let load balancer know you're draining.
•Wait for active requests to complete — Give in-flight operations time to finish.
•Wait for connections to return to pool — All checked-out connections should be returned.
•Close pool connections to database — Proper TCP termination for each connection.
•Shutdown complete — Process can exit safely.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import signal
import sys
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
 
pool: asyncpg.Pool = None
shutdown_event = asyncio.Event()
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan manager with graceful shutdown."""
    global pool
    
    # Startup
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@db:5432/app",
        min_size=5,
        max_size=20,
        max_inactive_connection_lifetime=300
    )
    print("Database pool initialized")
    
    yield
    
    # Shutdown (this block runs on SIGTERM/SIGINT)
    print("Starting graceful shutdown...")
    
    # Wait for active connections to complete (with timeout)
    shutdown_timeout = 30
    try:
        async with asyncio.timeout(shutdown_timeout):
            while pool.get_size() > pool.get_idle_size():
                print(f"Waiting for {pool.get_size() - pool.get_idle_size()} active connections...")
                await asyncio.sleep(1)
    except asyncio.TimeoutError:
        print(f"Shutdown timeout after {shutdown_timeout}s, forcing close")
    
    # Close pool
    await pool.close()
    print("Database pool closed")
 
app = FastAPI(lifespan=lifespan)
 
# Additional signal handling for robustness
def setup_signal_handlers():
    def signal_handler(sig, frame):
        print(f"Received signal {sig}, initiating shutdown...")
        shutdown_event.set()
    
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
const { Pool } = require('pg');
 
let pool = null;
let shuttingDown = false;
 
async function initializePool() {
  pool = new Pool({
    host: process.env.DB_HOST,
    database: process.env.DB_NAME,
    max: 20,
    idleTimeoutMillis: 30000,
  });
  
  // Test connection
  const client = await pool.connect();
  client.release();
  console.log('Database pool initialized');
}
 
async function gracefulShutdown(signal) {
  if (shuttingDown) return;
  shuttingDown = true;
  
  console.log(`Received ${signal}. Starting graceful shutdown...`);
  
  // Give in-flight requests time to complete
  const shutdownTimeout = 30000;
  const shutdownStart = Date.now();
  
  // Wait for active connections to drain
  while (pool.waitingCount > 0 || pool.totalCount > pool.idleCount) {
    if (Date.now() - shutdownStart > shutdownTimeout) {
      console.log('Shutdown timeout reached, forcing close');
      break;
    }
    console.log(`Waiting... Active: ${pool.totalCount - pool.idleCount}, Waiting: ${pool.waitingCount}`);
    await new Promise(resolve => setTimeout(resolve, 1000));
  }
  
  // Close pool
  await pool.end();
  console.log('Database pool closed');
  
  process.exit(0);
}
 
// Register shutdown handlers
process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
process.on('SIGINT', () => gracefulShutdown('SIGINT'));
 
// Ensure uncaught errors don't leave pool in bad state
process.on('uncaughtException', async (err) => {
  console.error('Uncaught exception:', err);
  await gracefulShutdown('uncaughtException');
});

Kubernetes Shutdown Grace Period

In Kubernetes, set terminationGracePeriodSeconds (default 30s) longer than your shutdown timeout. When Kubernetes sends SIGTERM, your app has this many seconds to shut down before SIGKILL. Ensure your connection drain + pool close completes within this window.

Testing Pool-Dependent Code

Testing code that uses connection pools requires strategies for isolation, speed, and reliability. Here are proven patterns.

Strategy 1: Test Database with Transactions

Run each test in a transaction that's rolled back after the test.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import pytest
 
@pytest.fixture(scope="session")
def db_pool():
    """Session-scoped pool for all tests."""
    pool = create_pool(TEST_DATABASE_URL)
    yield pool
    pool.close()
 
@pytest.fixture
def db_transaction(db_pool):
    """Transaction that's rolled back after each test."""
    conn = db_pool.getconn()
    conn.autocommit = False
    
    # Create savepoint for nested transaction support
    cursor = conn.cursor()
    cursor.execute("BEGIN")
    
    yield conn
    
    # Always rollback - test data never persists
    cursor.execute("ROLLBACK")
    db_pool.putconn(conn)
 
def test_create_user(db_transaction):
    conn = db_transaction
    cursor = conn.cursor()
    
    cursor.execute(
        "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id",
        ("Test User", "test@example.com")
    )
    user_id = cursor.fetchone()[0]
    
    cursor.execute("SELECT name FROM users WHERE id = %s", (user_id,))
    assert cursor.fetchone()[0] == "Test User"
    
    # Rollback happens automatically - no cleanup needed

Strategy 2: Mock Pool for Unit Tests

For true unit tests, mock the pool entirely.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from unittest.mock import Mock, MagicMock
from contextlib import contextmanager
 
class MockPool:
    """Mock pool for unit testing."""
    
    def __init__(self):
        self.mock_connection = MagicMock()
        self.mock_cursor = MagicMock()
        self.mock_connection.cursor.return_value = self.mock_cursor
    
    @contextmanager
    def connection(self):
        yield self.mock_connection
    
    def configure_result(self, rows):
        """Configure mock to return specific rows."""
        self.mock_cursor.fetchall.return_value = rows
        self.mock_cursor.fetchone.return_value = rows[0] if rows else None
 
def test_user_repository():
    mock_pool = MockPool()
    mock_pool.configure_result([(1, "Alice", "alice@example.com")])
    
    repo = UserRepository(mock_pool)
    user = repo.get_user(1)
    
    assert user.name == "Alice"
    mock_pool.mock_cursor.execute.assert_called_with(
        "SELECT * FROM users WHERE id = %s",
        (1,)
    )

Strategy 3: Testcontainers for Integration Tests

Spin up real databases in Docker for each test run.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from testcontainers.postgres import PostgresContainer
import pytest
 
@pytest.fixture(scope="session")
def postgres():
    """Spin up PostgreSQL in Docker for test session."""
    with PostgresContainer("postgres:15-alpine") as postgres:
        # Apply migrations
        engine = create_engine(postgres.get_connection_url())
        apply_migrations(engine)
        yield postgres
 
@pytest.fixture
def clean_db(postgres):
    """Clean database state for each test."""
    engine = create_engine(postgres.get_connection_url())
    with engine.connect() as conn:
        # Truncate all tables
        conn.execute("TRUNCATE users, orders, products CASCADE")
        conn.commit()
    yield engine
 
def test_user_crud(clean_db):
    pool = create_pool(clean_db.url)
    repo = UserRepository(pool)
    
    # Create
    user = repo.create(name="Alice", email="alice@test.com")
    assert user.id is not None
    
    # Read
    fetched = repo.get_user(user.id)
    assert fetched.name == "Alice"
    
    pool.close()

Testing Recommendation

Use mocks for unit tests (fast, isolated), transaction rollback for integration tests (real DB, fast cleanup), and testcontainers for CI/CD (fully isolated, reproducible). This three-tier approach balances speed and confidence.

Troubleshooting Common Pool Issues

When connection pooling goes wrong, symptoms can be confusing. Here's a troubleshooting guide for the most common issues.

Pool Issue Diagnosis
Symptom	Likely Cause	Solution
Connection timeout errors	Pool exhausted, all connections in use	Check for leaks, reduce hold time, increase pool size
Slow first request after idle	Connection validation failing, creating new	Configure min connections, adjust validation strategy
Intermittent connection errors	Connections hitting max lifetime	Reduce max lifetime, add connection retry logic
Database connection count growing	Leaking connections (not returning to pool)	Audit all DB-using code, add leak detection
Sporadic query failures	Broken connections not detected	Enable test-on-borrow, add connection validation
High memory usage	Too many connections, each consuming memory	Reduce pool size, review actual concurrency needs

Leak detection techniques:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import traceback
import time
import threading
from collections import defaultdict
 
class LeakDetectingPool:
    """Pool wrapper that tracks and reports connection leaks."""
    
    def __init__(self, pool, leak_threshold_seconds=60):
        self._pool = pool
        self._leak_threshold = leak_threshold_seconds
        self._checkouts = {}  # conn_id -> (checkout_time, stack_trace)
        self._lock = threading.Lock()
        
        # Start leak detection thread
        self._monitor_thread = threading.Thread(target=self._monitor_leaks, daemon=True)
        self._monitor_thread.start()
    
    def getconn(self):
        conn = self._pool.getconn()
        
        with self._lock:
            self._checkouts[id(conn)] = (
                time.time(),
                traceback.format_stack()
            )
        
        return conn
    
    def putconn(self, conn):
        with self._lock:
            if id(conn) in self._checkouts:
                del self._checkouts[id(conn)]
        
        self._pool.putconn(conn)
    
    def _monitor_leaks(self):
        while True:
            time.sleep(30)
            self._check_leaks()
    
    def _check_leaks(self):
        now = time.time()
        
        with self._lock:
            leaks = []
            for conn_id, (checkout_time, stack) in self._checkouts.items():
                hold_time = now - checkout_time
                if hold_time > self._leak_threshold:
                    leaks.append((hold_time, stack))
        
        for hold_time, stack in leaks:
            print(f"POTENTIAL CONNECTION LEAK: held for {hold_time:.0f}s")
            print("Checkout location:")
            print("".join(stack[-10:]))  # Last 10 frames

Key Takeaways

•Use context managers everywhere — Automatic connection return prevents leaks.
•Keep transactions short — Minimize connection hold time for pool efficiency.
•Implement graceful shutdown — Drain connections before closing pool.
•Pool should be singleton — One pool instance, shared across entire application.
•Handle errors explicitly — Every error path must return connections to pool.
•Enable leak detection — In development and staging, catch leaks early.
•Test with real databases — Use testcontainers for reliable integration tests.
•Monitor pool metrics — Track checkout times, wait times, utilization.

Module Complete

You now have comprehensive knowledge of connection pooling—from the fundamental reasons why it matters, through sizing strategies, implementation options across languages, external poolers like PgBouncer and ProxySQL, and advanced application-side patterns. Connection management is foundational to building database applications that scale reliably.

5 / 5

Loading learning content...

System Design (HLD)Connection Pooling

Connection Pooling: Managing Database Connections at Scale

LevelIntermediate

Duration75 mins

TopicConnection Pooling

5 / 5

Application-side Pooling

Mastering Connection Management in Your Application

What You Will Learn

The Application Pool Architecture

Core pool components:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
┌─────────────────────────────────────────────────────────────────┐
│                    Application Process                          │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │                  Connection Pool                        │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Idle Connection Queue              │   │   │
│  │  │  [Conn1] [Conn2] [Conn3] [Conn4] [Conn5]       │   │   │
│  │  │  Ready for immediate checkout                   │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Active Connections                 │   │   │
│  │  │  [Conn6 → Request A] [Conn7 → Request B]       │   │   │
│  │  │  Currently executing queries                    │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Wait Queue (if pool exhausted)      │   │   │
│  │  │  [Request C waiting] [Request D waiting]        │   │   │
│  │  │  FIFO queue for fairness                        │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  │  ┌─────────────────────────────────────────────────┐   │   │
│  │  │              Background Services                │   │   │
│  │  │  • Connection validator (periodic health check)  │   │   │
│  │  │  • Connection killer (max lifetime enforcement) │   │   │
│  │  │  • Pool scaler (min/max adjustment)             │   │   │
│  │  │  • Metrics collector                            │   │   │
│  │  └─────────────────────────────────────────────────┘   │   │
│  │                                                         │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Connection lifecycle states:

Connection States in Application Pool
State	Description	Transitions
Creating	Connection being established to database	→ Idle (success) or Destroyed (failure)
Idle	In pool, ready for checkout	→ Active (checkout) or Validating (health check)
Validating	Being tested for health	→ Idle (valid) or Destroyed (invalid)
Active	Checked out, executing queries	→ Idle (return) or Destroyed (error/leak)
Retiring	Past max lifetime, waiting to close	→ Destroyed (when returned or timeout)
Destroyed	Closed, removed from pool	New connection Created to replace

The Validation Tradeoff

Connection Checkout Patterns

How you acquire connections from the pool significantly impacts reliability and performance. There are several patterns to consider.

Pattern 1: Simple Checkout/Return

The basic pattern where the calling code explicitly manages connection lifecycle.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Explicit checkout/return - requires discipline
class UserRepository:
    def __init__(self, pool):
        self.pool = pool
    
    def get_user(self, user_id: int) -> User:
        conn = self.pool.getconn()      # Checkout
        try:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT * FROM users WHERE id = %s", 
                (user_id,)
            )
            row = cursor.fetchone()
            return User.from_row(row) if row else None
        finally:
            self.pool.putconn(conn)     # Always return!

Pattern 2: Context Manager Pattern

Automatic resource management using language-specific constructs.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from contextlib import contextmanager
 
class DatabasePool:
    def __init__(self, connection_string):
        self._pool = psycopg_pool.ConnectionPool(connection_string)
    
    @contextmanager
    def connection(self):
        """Automatic checkout/return with context manager."""
        conn = self._pool.getconn()
        try:
            yield conn
            conn.commit()       # Auto-commit on success
        except Exception:
            conn.rollback()     # Auto-rollback on error
            raise
        finally:
            self._pool.putconn(conn)   # Always return
 
# Usage - impossible to leak connections
def get_user(db: DatabasePool, user_id: int) -> User:
    with db.connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        row = cursor.fetchone()
        return User.from_row(row) if row else None
    # Connection automatically returned here

Pattern 3: Decorator Pattern

Wrap functions with automatic connection injection.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from functools import wraps
 
# Global pool instance (initialized at startup)
db_pool: DatabasePool = None
 
def with_connection(func):
    """Inject connection as first argument."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        with db_pool.connection() as conn:
            return func(conn, *args, **kwargs)
    return wrapper
 
@with_connection
def get_user(conn, user_id: int) -> User:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    row = cursor.fetchone()
    return User.from_row(row) if row else None
 
# Caller doesn't manage connection at all
user = get_user(123)

Pattern 4: Request-Scoped Pattern

Connection shared across a single request, common in web frameworks.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from fastapi import Depends, FastAPI
from contextlib import asynccontextmanager
 
app = FastAPI()
pool: asyncpg.Pool = None
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool
    pool = await asyncpg.create_pool(
        "postgresql://user:pass@db:5432/myapp",
        min_size=5,
        max_size=20
    )
    yield
    await pool.close()
 
app = FastAPI(lifespan=lifespan)
 
async def get_connection():
    """Dependency injection for request-scoped connection."""
    async with pool.acquire() as conn:
        yield conn
 
@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    conn = Depends(get_connection)  # Injected, auto-returned
):
    row = await conn.fetchrow(
        "SELECT * FROM users WHERE id = $1", 
        user_id
    )
    return dict(row) if row else None

One Connection Per Request

Error Handling and Recovery

Database connections fail. Networks partition. Servers restart. Your application must handle these failures gracefully without leaking connections or corrupting state.

Classifying connection errors:

Connection Error Types and Handling
Error Type	Examples	Recovery Strategy
Transient	Network glitch, server briefly unavailable	Retry with backoff, discard connection
Connection broken	Server restart, firewall timeout, network change	Discard connection, create new one
Pool exhausted	All connections in use, wait queue full	Fail fast with clear error, don't block
Authentication	Invalid credentials, permission revoked	Fail immediately, alert operators
Query/Application	Syntax error, constraint violation	Return connection to pool, propagate error

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import time
import random
from contextlib import contextmanager
 
class PoolExhaustedError(Exception):
    """Raised when pool cannot provide connection in time."""
    pass
 
class DatabasePool:
    def __init__(self, pool, max_retries=3, base_delay=0.1):
        self._pool = pool
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    @contextmanager
    def connection(self, retry=True):
        """Get connection with error handling and optional retry."""
        conn = None
        last_error = None
        
        for attempt in range(self.max_retries if retry else 1):
            try:
                # Attempt to get connection with timeout
                conn = self._pool.getconn(timeout=5.0)
                
                # Validate connection before use
                self._validate_connection(conn)
                
                try:
                    yield conn
                    conn.commit()  # Commit on success
                    return
                except Exception as e:
                    self._handle_query_error(conn, e)
                    raise
                    
            except psycopg.PoolTimeout:
                raise PoolExhaustedError(
                    "Could not acquire connection from pool within timeout"
                )
            except psycopg.OperationalError as e:
                last_error = e
                if conn:
                    self._discard_connection(conn)
                    conn = None
                
                if attempt < self.max_retries - 1:
                    delay = self.base_delay * (2 ** attempt) + random.uniform(0, 0.1)
                    time.sleep(delay)
                    continue
                raise
            finally:
                if conn:
                    try:
                        self._pool.putconn(conn)
                    except Exception:
                        pass  # Connection already broken/closed
    
    def _validate_connection(self, conn):
        """Quick health check before use."""
        try:
            cursor = conn.cursor()
            cursor.execute("SELECT 1")
            cursor.fetchone()
        except Exception:
            raise psycopg.OperationalError("Connection validation failed")
    
    def _handle_query_error(self, conn, error):
        """Handle errors during query execution."""
        try:
            conn.rollback()
        except Exception:
            # Connection broken during rollback - mark for discard
            self._discard_connection(conn)
    
    def _discard_connection(self, conn):
        """Mark connection as broken so pool discards it."""
        # Implementation depends on pool library
        # Most pools detect broken connections on return
        pass

Never Swallow Connection Errors

Transaction Management

Transactions and connection pooling interact in subtle ways. Managing transactions correctly is essential for data integrity and pool efficiency.

The transaction-connection binding:

Transaction-Pool Implications

•Long transactions hold connections — A 30-second transaction keeps that connection unavailable for 30 seconds, reducing effective pool size.
•Transaction leaks are pool leaks — Starting a transaction and not completing it (via exception without rollback) leaves the connection unusable.
•Autocommit for reads — Read-only queries without explicit transactions should use autocommit mode for maximum pool efficiency.
•Transaction scope should be minimal — Do non-database work outside transactions to minimize connection hold time.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
from contextlib import contextmanager
 
class TransactionManager:
    def __init__(self, pool):
        self.pool = pool
    
    @contextmanager
    def transaction(self, isolation_level="READ COMMITTED"):
        """Explicit transaction with guaranteed cleanup."""
        conn = self.pool.getconn()
        conn.set_session(isolation_level=isolation_level, autocommit=False)
        
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.set_session(autocommit=True)  # Reset to autocommit
            self.pool.putconn(conn)
    
    @contextmanager
    def read_only(self):
        """Optimized for read-only operations."""
        conn = self.pool.getconn()
        conn.set_session(readonly=True, autocommit=True)
        
        try:
            yield conn
        finally:
            conn.set_session(readonly=False)
            self.pool.putconn(conn)
 
# Usage
def transfer_money(tx_manager, from_id, to_id, amount):
    with tx_manager.transaction(isolation_level="SERIALIZABLE") as conn:
        cursor = conn.cursor()
        
        # Debit source account
        cursor.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_id)
        )
        
        # Credit destination account
        cursor.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_id)
        )
        
        # Verify source had sufficient balance
        cursor.execute(
            "SELECT balance FROM accounts WHERE id = %s",
            (from_id,)
        )
        if cursor.fetchone()[0] < 0:
            raise ValueError("Insufficient funds")
        
        # Transaction commits automatically via context manager
 
def get_account_summary(tx_manager, account_id):
    with tx_manager.read_only() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT balance, last_transaction FROM accounts WHERE id = %s",
            (account_id,)
        )
        return cursor.fetchone()

Keep Transactions Short

Dependency Injection Integration

Modern applications use dependency injection for clean architecture and testability. Integrating connection pools with DI requires careful consideration of scopes and lifecycle.

Java (Spring Framework):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@Configuration
public class DatabaseConfig {
    
    @Bean
    @ConfigurationProperties("spring.datasource.hikari")
    public HikariConfig hikariConfig() {
        return new HikariConfig();
    }
    
    @Bean(destroyMethod = "close")
    public DataSource dataSource(HikariConfig config) {
        // Pool created as singleton, shared across entire application
        return new HikariDataSource(config);
    }
    
    @Bean
    public JdbcTemplate jdbcTemplate(DataSource dataSource) {
        return new JdbcTemplate(dataSource);
    }
    
    @Bean
    public TransactionManager transactionManager(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }
}
 
// Repository using injected JdbcTemplate
@Repository
public class UserRepository {
    private final JdbcTemplate jdbc;
    
    public UserRepository(JdbcTemplate jdbc) {
        this.jdbc = jdbc;  // Constructor injection
    }
    
    public User findById(Long id) {
        // JdbcTemplate handles connection checkout/return internally
        return jdbc.queryForObject(
            "SELECT * FROM users WHERE id = ?",
            new UserRowMapper(),
            id
        );
    }
}

Node.js (NestJS):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import { Module, Global, OnApplicationShutdown } from '@nestjs/common';
import { Pool } from 'pg';
 
const DATABASE_POOL = 'DATABASE_POOL';
 
@Global()
@Module({
  providers: [
    {
      provide: DATABASE_POOL,
      useFactory: async () => {
        const pool = new Pool({
          host: process.env.DB_HOST,
          port: parseInt(process.env.DB_PORT),
          database: process.env.DB_NAME,
          user: process.env.DB_USER,
          password: process.env.DB_PASSWORD,
          max: 20,
          idleTimeoutMillis: 30000,
          connectionTimeoutMillis: 10000,
        });
        
        // Validate pool is working
        const client = await pool.connect();
        client.release();
        
        return pool;
      },
    },
  ],
  exports: [DATABASE_POOL],
})
export class DatabaseModule implements OnApplicationShutdown {
  constructor(
    @Inject(DATABASE_POOL) private readonly pool: Pool
  ) {}
  
  async onApplicationShutdown(signal?: string) {
    await this.pool.end();
    console.log('Database pool closed');
  }
}
 
// Service using injected pool
@Injectable()
export class UserService {
  constructor(
    @Inject(DATABASE_POOL) private readonly pool: Pool
  ) {}
  
  async findById(id: number): Promise<User | null> {
    const result = await this.pool.query(
      'SELECT * FROM users WHERE id = $1',
      [id]
    );
    return result.rows[0] || null;
  }
}

Pool as Singleton

Graceful Shutdown

When your application stops—for deployment, scaling, or failure—it must close database connections properly. Improper shutdown leaves connections orphaned on the database server.

The graceful shutdown sequence:

Shutdown Steps

•Stop accepting new requests — Let load balancer know you're draining.
•Wait for active requests to complete — Give in-flight operations time to finish.
•Wait for connections to return to pool — All checked-out connections should be returned.
•Close pool connections to database — Proper TCP termination for each connection.
•Shutdown complete — Process can exit safely.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import signal
import sys
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI
 
pool: asyncpg.Pool = None
shutdown_event = asyncio.Event()
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan manager with graceful shutdown."""
    global pool
    
    # Startup
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@db:5432/app",
        min_size=5,
        max_size=20,
        max_inactive_connection_lifetime=300
    )
    print("Database pool initialized")
    
    yield
    
    # Shutdown (this block runs on SIGTERM/SIGINT)
    print("Starting graceful shutdown...")
    
    # Wait for active connections to complete (with timeout)
    shutdown_timeout = 30
    try:
        async with asyncio.timeout(shutdown_timeout):
            while pool.get_size() > pool.get_idle_size():
                print(f"Waiting for {pool.get_size() - pool.get_idle_size()} active connections...")
                await asyncio.sleep(1)
    except asyncio.TimeoutError:
        print(f"Shutdown timeout after {shutdown_timeout}s, forcing close")
    
    # Close pool
    await pool.close()
    print("Database pool closed")
 
app = FastAPI(lifespan=lifespan)
 
# Additional signal handling for robustness
def setup_signal_handlers():
    def signal_handler(sig, frame):
        print(f"Received signal {sig}, initiating shutdown...")
        shutdown_event.set()
    
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
const { Pool } = require('pg');
 
let pool = null;
let shuttingDown = false;
 
async function initializePool() {
  pool = new Pool({
    host: process.env.DB_HOST,
    database: process.env.DB_NAME,
    max: 20,
    idleTimeoutMillis: 30000,
  });
  
  // Test connection
  const client = await pool.connect();
  client.release();
  console.log('Database pool initialized');
}
 
async function gracefulShutdown(signal) {
  if (shuttingDown) return;
  shuttingDown = true;
  
  console.log(`Received ${signal}. Starting graceful shutdown...`);
  
  // Give in-flight requests time to complete
  const shutdownTimeout = 30000;
  const shutdownStart = Date.now();
  
  // Wait for active connections to drain
  while (pool.waitingCount > 0 || pool.totalCount > pool.idleCount) {
    if (Date.now() - shutdownStart > shutdownTimeout) {
      console.log('Shutdown timeout reached, forcing close');
      break;
    }
    console.log(`Waiting... Active: ${pool.totalCount - pool.idleCount}, Waiting: ${pool.waitingCount}`);
    await new Promise(resolve => setTimeout(resolve, 1000));
  }
  
  // Close pool
  await pool.end();
  console.log('Database pool closed');
  
  process.exit(0);
}
 
// Register shutdown handlers
process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
process.on('SIGINT', () => gracefulShutdown('SIGINT'));
 
// Ensure uncaught errors don't leave pool in bad state
process.on('uncaughtException', async (err) => {
  console.error('Uncaught exception:', err);
  await gracefulShutdown('uncaughtException');
});

Kubernetes Shutdown Grace Period

Testing Pool-Dependent Code

Testing code that uses connection pools requires strategies for isolation, speed, and reliability. Here are proven patterns.

Strategy 1: Test Database with Transactions

Run each test in a transaction that's rolled back after the test.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import pytest
 
@pytest.fixture(scope="session")
def db_pool():
    """Session-scoped pool for all tests."""
    pool = create_pool(TEST_DATABASE_URL)
    yield pool
    pool.close()
 
@pytest.fixture
def db_transaction(db_pool):
    """Transaction that's rolled back after each test."""
    conn = db_pool.getconn()
    conn.autocommit = False
    
    # Create savepoint for nested transaction support
    cursor = conn.cursor()
    cursor.execute("BEGIN")
    
    yield conn
    
    # Always rollback - test data never persists
    cursor.execute("ROLLBACK")
    db_pool.putconn(conn)
 
def test_create_user(db_transaction):
    conn = db_transaction
    cursor = conn.cursor()
    
    cursor.execute(
        "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id",
        ("Test User", "test@example.com")
    )
    user_id = cursor.fetchone()[0]
    
    cursor.execute("SELECT name FROM users WHERE id = %s", (user_id,))
    assert cursor.fetchone()[0] == "Test User"
    
    # Rollback happens automatically - no cleanup needed

Strategy 2: Mock Pool for Unit Tests

For true unit tests, mock the pool entirely.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from unittest.mock import Mock, MagicMock
from contextlib import contextmanager
 
class MockPool:
    """Mock pool for unit testing."""
    
    def __init__(self):
        self.mock_connection = MagicMock()
        self.mock_cursor = MagicMock()
        self.mock_connection.cursor.return_value = self.mock_cursor
    
    @contextmanager
    def connection(self):
        yield self.mock_connection
    
    def configure_result(self, rows):
        """Configure mock to return specific rows."""
        self.mock_cursor.fetchall.return_value = rows
        self.mock_cursor.fetchone.return_value = rows[0] if rows else None
 
def test_user_repository():
    mock_pool = MockPool()
    mock_pool.configure_result([(1, "Alice", "alice@example.com")])
    
    repo = UserRepository(mock_pool)
    user = repo.get_user(1)
    
    assert user.name == "Alice"
    mock_pool.mock_cursor.execute.assert_called_with(
        "SELECT * FROM users WHERE id = %s",
        (1,)
    )

Strategy 3: Testcontainers for Integration Tests

Spin up real databases in Docker for each test run.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from testcontainers.postgres import PostgresContainer
import pytest
 
@pytest.fixture(scope="session")
def postgres():
    """Spin up PostgreSQL in Docker for test session."""
    with PostgresContainer("postgres:15-alpine") as postgres:
        # Apply migrations
        engine = create_engine(postgres.get_connection_url())
        apply_migrations(engine)
        yield postgres
 
@pytest.fixture
def clean_db(postgres):
    """Clean database state for each test."""
    engine = create_engine(postgres.get_connection_url())
    with engine.connect() as conn:
        # Truncate all tables
        conn.execute("TRUNCATE users, orders, products CASCADE")
        conn.commit()
    yield engine
 
def test_user_crud(clean_db):
    pool = create_pool(clean_db.url)
    repo = UserRepository(pool)
    
    # Create
    user = repo.create(name="Alice", email="alice@test.com")
    assert user.id is not None
    
    # Read
    fetched = repo.get_user(user.id)
    assert fetched.name == "Alice"
    
    pool.close()

Testing Recommendation

Troubleshooting Common Pool Issues

When connection pooling goes wrong, symptoms can be confusing. Here's a troubleshooting guide for the most common issues.

Pool Issue Diagnosis
Symptom	Likely Cause	Solution
Connection timeout errors	Pool exhausted, all connections in use	Check for leaks, reduce hold time, increase pool size
Slow first request after idle	Connection validation failing, creating new	Configure min connections, adjust validation strategy
Intermittent connection errors	Connections hitting max lifetime	Reduce max lifetime, add connection retry logic
Database connection count growing	Leaking connections (not returning to pool)	Audit all DB-using code, add leak detection
Sporadic query failures	Broken connections not detected	Enable test-on-borrow, add connection validation
High memory usage	Too many connections, each consuming memory	Reduce pool size, review actual concurrency needs

Leak detection techniques:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import traceback
import time
import threading
from collections import defaultdict
 
class LeakDetectingPool:
    """Pool wrapper that tracks and reports connection leaks."""
    
    def __init__(self, pool, leak_threshold_seconds=60):
        self._pool = pool
        self._leak_threshold = leak_threshold_seconds
        self._checkouts = {}  # conn_id -> (checkout_time, stack_trace)
        self._lock = threading.Lock()
        
        # Start leak detection thread
        self._monitor_thread = threading.Thread(target=self._monitor_leaks, daemon=True)
        self._monitor_thread.start()
    
    def getconn(self):
        conn = self._pool.getconn()
        
        with self._lock:
            self._checkouts[id(conn)] = (
                time.time(),
                traceback.format_stack()
            )
        
        return conn
    
    def putconn(self, conn):
        with self._lock:
            if id(conn) in self._checkouts:
                del self._checkouts[id(conn)]
        
        self._pool.putconn(conn)
    
    def _monitor_leaks(self):
        while True:
            time.sleep(30)
            self._check_leaks()
    
    def _check_leaks(self):
        now = time.time()
        
        with self._lock:
            leaks = []
            for conn_id, (checkout_time, stack) in self._checkouts.items():
                hold_time = now - checkout_time
                if hold_time > self._leak_threshold:
                    leaks.append((hold_time, stack))
        
        for hold_time, stack in leaks:
            print(f"POTENTIAL CONNECTION LEAK: held for {hold_time:.0f}s")
            print("Checkout location:")
            print("".join(stack[-10:]))  # Last 10 frames

Key Takeaways

•Use context managers everywhere — Automatic connection return prevents leaks.
•Keep transactions short — Minimize connection hold time for pool efficiency.
•Implement graceful shutdown — Drain connections before closing pool.
•Pool should be singleton — One pool instance, shared across entire application.
•Handle errors explicitly — Every error path must return connections to pool.
•Enable leak detection — In development and staging, catch leaks early.
•Test with real databases — Use testcontainers for reliable integration tests.
•Monitor pool metrics — Track checkout times, wait times, utilization.

Module Complete

5 / 5