How to Build a Connection Pool Reuse Mock in Python

Build a mock connection pool with context manager to track connection reuse, acquires, and releases in Python.

Medium Python 3.9+ Aug 9, 2026 Database scaling & optimization 13 views 0 copies

Python code

65 lines
Python 3.9+
import time
from contextlib import contextmanager


class Connection:
    def __init__(self, name):
        self.name = name
        self.in_use = False
        self.busy_since = None

    def fetch(self):
        return f"data from {self.name}"


class ConnectionPool:
    def __init__(self, size=3):
        self.connections = [Connection(f"conn-{i}") for i in range(size)]
        self.metrics = {"acquires": 0, "reuses": 0, "releases": 0}

    @contextmanager
    def get_connection(self):
        conn = self._acquire()
        try:
            yield conn
        finally:
            self._release(conn)

    def _acquire(self):
        # Try to reuse an idle connection first
        for conn in self.connections:
            if not conn.in_use:
                conn.in_use = True
                conn.busy_since = time.time()
                if any(c.in_use for c in self.connections):
                    self.metrics["reuses"] += 1
                self.metrics["acquires"] += 1
                return conn
        # No idle connection — create a new mock connection
        new_conn = Connection(f"conn-{len(self.connections)}")
        new_conn.in_use = True
        new_conn.busy_since = time.time()
        self.connections.append(new_conn)
        self.metrics["acquires"] += 1
        return new_conn

    def _release(self, conn):
        conn.in_use = False
        conn.busy_since = None
        self.metrics["releases"] += 1


if __name__ == "__main__":
    pool = ConnectionPool(size=2)

    with pool.get_connection() as c1:
        print(c1.fetch())
        with pool.get_connection() as c2:
            print(c2.fetch())
        # c1 still in use

    # After exiting first with-block, c1 is free for reuse
    with pool.get_connection() as c3:
        print(c3.fetch())

    print("Metrics:", pool.metrics)

Output

stdout
data from conn-0
data from conn-1
data from conn-0
Metrics: {'acquires': 3, 'reuses': 1, 'releases': 3}

How it works

The ConnectionPool class maintains a fixed set of mock connections and uses a context manager to ensure every acquired connection is always released, even when an exception is raised. The _acquire method first scans for any idle connection; if one exists it marks it as busy and increments the reuse metric, otherwise it grows the pool by appending a new connection. This mirrors how real database pools (like those in SQLAlchemy or psycopg2) manage thresholds for max connections. Metrics tracking acquires, reuses, and releases lets you verify pool efficiency — in this example the third acquisition reuses conn-0 after it was released.

Common mistakes

  • Forgetting to release connections in a `finally` block, causing pool exhaustion under exceptions
  • Counting every acquire as a reuse, even when no idle connection actually existed
  • Not setting `busy_since` on reused connections, breaking timeout/Age logic

Variations

  1. Use `queue.Queue` to store idle connections and pop them from the front for strict FIFO ordering
  2. Add max-pool-size guard that raises or waits instead of growing the pool indefinitely
  3. Use `threading.local()` to track per-thread connections for thread-safe reuse

Real-world use cases

  • Verifying connection reuse in a web app before switching from raw psycopg2 to SQLAlchemy's built-in pool.
  • Unit-testing custom pool logic like max-connections limits or idle-timeout eviction without a live database.
  • Profiling connection churn in a backend service to spot code paths that leak or over-acquire DB connections.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Database scaling & optimization

Related tutorials and quizzes for this topic.