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:

┌─────────────────────────────────────────────────────────────────┐
│                    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.

5 / 5

┌─────────────────────────────────────────────────────────────────┐ │ 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 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

# 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!

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

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)

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

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

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

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()

@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 ); } }

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; } }

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)

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'); });

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

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,) )

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()

Application-side Pooling

Mastering Connection Management in Your Application

The Application Pool Architecture

Connection Checkout Patterns

Error Handling and Recovery

Transaction Management

Dependency Injection Integration

Graceful Shutdown

Testing Pool-Dependent Code

There’s more to understand.

Application-side Pooling

Mastering Connection Management in Your Application

The Application Pool Architecture

Connection Checkout Patterns

Error Handling and Recovery

Transaction Management

Dependency Injection Integration

Graceful Shutdown

Testing Pool-Dependent Code

There’s more to understand.