Object Pool Pattern for Database Connections in Python
Implements a reusable connection pool with acquire/release and context manager support, mocking database connections with idle reuse and exhaustion handling.
Python code
53 linesimport time
from contextlib import contextmanager
from collections import deque
class ConnectionPool:
def __init__(self, size=3, max_idle=5):
self._idle = deque(maxlen=max_idle)
self._active = set()
self.size = size
def _create(self):
return {"created_at": time.time(), "queries": 0}
def acquire(self):
if self._idle:
conn = self._idle.popleft()
elif len(self._active) < self.size:
conn = self._create()
else:
raise RuntimeError("Pool exhausted")
self._active.add(id(conn))
return conn
def release(self, conn):
self._active.discard(id(conn))
conn["queries"] += 1
self._idle.append(conn)
@contextmanager
def connection(self):
conn = self.acquire()
try:
yield conn
finally:
self.release(conn)
if __name__ == "__main__":
pool = ConnectionPool(size=2)
with pool.connection() as c1:
print("Acquired:", c1)
with pool.connection() as c2:
print("Acquired:", c2)
try:
pool.acquire()
except RuntimeError as e:
print("Error:", e)
with pool.connection() as c3:
print("Reused idle:", c3)
print("Idle queue size:", len(pool._idle))
Output
Acquired: {'created_at': 1710000000.123, 'queries': 0}
Acquired: {'created_at': 1710000000.456, 'queries': 0}
Error: Pool exhausted
Reused idle: {'created_at': 1710000000.456, 'queries': 1}
Idle queue size: 1
How it works
The pool maintains idle connections in a deque with a max length, and tracks active connections by their id(). On acquire, it pops from idle first, then creates new ones if under the size limit, otherwise raises RuntimeError. release increments a query counter and pushes the connection back to idle. The context manager guarantees release even on exceptions via finally. Each connection is a mock dict, but the pattern maps directly to real DB driver objects like psycopg2 or mysql.connector.
Common mistakes
- Releasing the same connection twice, corrupting idle pool state
- Forgetting to release connections on exception paths without context managers
- Not binding pool size to actual database connection limits, causing over-allocation
Variations
- Use thread-local pools with `threading.local()` for per-thread isolation
- Add validation on release to discard broken connections instead of returning them to idle
Real-world use cases
- Managing PostgreSQL connections in a web API to avoid connection setup overhead per request.
- Reusing expensive MySQL connections in a batch ETL job processing millions of rows.
- Pooling Redis client connections in a rate-limiter service to handle concurrent requests efficiently.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States medium
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
Keep learning
Related tutorials and quizzes for this topic.