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.
Python code
65 linesimport 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
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
- Use `queue.Queue` to store idle connections and pop them from the front for strict FIFO ordering
- Add max-pool-size guard that raises or waits instead of growing the pool indefinitely
- 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
More from Database scaling & optimization
- Approximate Count with HyperLogLog in Python medium
- B-Tree Insert and In-Order Traversal in Python hard
- Broadcast a Small Reference Table in Python easy
- Build a Full Text Search Index in Python medium
- Build a Partial Index Mock in Python for Database Filtering easy
- Composite index leftmost prefix in Python medium
Keep learning
Related tutorials and quizzes for this topic.