Implement a Connection Pool
Company: Airtable
Role: Software Engineer
Category: Software Engineering Fundamentals
Difficulty: medium
Interview Round: Technical Screen
You are given an existing `DBConnection` class that simulates a physical database connection. It exposes the following methods and must not be modified:
- `initialize()`
- `query(sql: str)`
- `close()`
Implement the missing behavior for `ConnectionPool` and `PoolConnection`.
Skeleton:
`class DBConnection:`
- `__init__(self)`
- `initialize(self)`
- `query(self, sql: str) -> str`
- `close(self)`
Destroys the physical connection.
`class ConnectionPool:`
- `__init__(self, max_conn: int)`
- `get_connection(self) -> PoolConnection | None`
`class PoolConnection:`
- `__init__(self, parent: ConnectionPool, db_conn: DBConnection)`
- `query(self, sql: str)`
- `close(self)`
Requirements:
1. `ConnectionPool(max_conn)`
- Throw an error if `max_conn <= 0`.
- Track how many connections are currently checked out.
- Track previously used `DBConnection` instances that have been returned to the pool.
2. `ConnectionPool.get_connection()`
- If the number of active checked-out connections has reached `max_conn`, return `None`.
- If an idle `DBConnection` is available in the pool, reuse it.
- Otherwise create a new `DBConnection`, initialize it, and return it wrapped in a `PoolConnection`.
3. `PoolConnection.query(sql)`
- Forward the call to the underlying `DBConnection.query(sql)`.
- If this `PoolConnection` has already been closed, throw an error.
4. `PoolConnection.close()`
- Return the underlying `DBConnection` to the parent `ConnectionPool` so it can be reused later.
- Do not destroy the underlying `DBConnection` when a `PoolConnection` is closed.
- Prevent a closed `PoolConnection` from being queried again.
- Prevent double-closing, or treat it as an error.
The goal is to implement a simple reusable connection pool with correct ownership and lifecycle handling.
Overview: This question evaluates a candidate's competence in resource management, object ownership and lifecycle handling, and safe reuse of pooled database connections.
Community answers
Answer by aniruddhachowdhuru898
class ConnectionPool:
def init(self, max_conn: int):
if max_conn <= 0:
raise ValueError("max_conn must be greater than 0")
self.max_conn = max_conn
self.active_connections = 0
self.idle_connections = []
def get_connection(self) -> "PoolConnection | None":
# Enforce the maximum number of checked-out connections.
if self.active_connections >= self.max_conn:
return None
# Reuse a previously returned physical connection.
if self.idle_connections:
db_conn = self.idle_connections.pop()
else:
# Create and initialize a new physical connection.
db_conn = DBConnection()
try:
db_conn.initialize()
except Exception:
# Do not count a failed initialization as active.
try:
db_conn.close()
except Exception:
pass
raise
self.active_connections += 1
return PoolConnection(self, db_conn)
def _release_connection(self, db_conn: "DBConnection"):
# Return the physical connection for future reuse.
self.idle_connections.append(db_conn)
self.active_connections -= 1
class PoolConnection:
def init(self, parent: ConnectionPool, db_conn: "DBConnection"):
self.parent = parent
self.db_conn = db_conn
self.closed = False
def query(self, sql: str):
if self.closed:
raise RuntimeError("Connection has already been closed")
return self.db_conn.query(sql)
def close(self):
if self.closed:
raise RuntimeError("Connection has already been closed")
self.closed = True
self.parent._release_connection(self.db_conn)