Simple Resource Management: Implementing the Object Pool Pattern in Python

Introduction

Sometimes, for the sake of efficiency, it can be really helpful to keep a group of ready-to-use objects, known as a “pool.” This can be particularly useful when dealing with expensive resources like database connections, which take up a lot of time and resources to create.

Here’s the basic idea: You create a pool of objects, usually stored in an array, and have a manager object responsible for handing out these objects when needed and keeping track of their usage.

Implementation in Python

In the following example, we’ll simulate a database connection pool. To ensure that our pool works safely with multiple threads, we start by importing the threading module.

import threading

Next, we’ll define a class called DbConnection that represents a basic database connection. Each connection has a unique ID.

class DbConnection:
    _id: int = None

    def __init__(self, initial_value: int):
        self._id = initial_value

    def get_id(self) -> int:
        return self._id

The core of our implementation is the ConnectionPool class, which manages the database connections. It has attributes for keeping track of idle and active connections, a capacity limit, and a lock to make sure the pattern is thread-safe.

class ConnectionPool:
    _idleConnections: list[DbConnection] = None
    _activeConnections: list[DbConnection] = None
    _capacity: int = None
    _poolLock: threading.Lock = threading.Lock()

    def __init__(self, initial_connections: list[DbConnection]):
        self._idleConnections = initial_connections
        self._activeConnections = []
        self._capacity = len(initial_connections)

    def get_connection(self) -> DbConnection | None:
        self._poolLock.acquire()
        try:
            if len(self._idleConnections) == 0:
                return None
            result: DbConnection = self._idleConnections[0]
            self._idleConnections = self._idleConnections[1:]
            self._activeConnections.append(result)
            print(f"Gave out connection with Id {result.get_id()}")
            return result
        finally:
            self._poolLock.release()

    def remove_from_active(self, connection: DbConnection) -> DbConnection | None:
        active_length = len(self._activeConnections)
        for index, conn in enumerate(self._activeConnections):
            if conn.get_id() == connection.get_id():
                self._activeConnections[active_length - 1], self._activeConnections[index] = self._activeConnections[
                    index], self._activeConnections[active_length - 1]
                self._activeConnections = self._activeConnections[:active_length - 1]
                return connection
        return None

    def return_connection(self, connection: DbConnection):
        self._poolLock.acquire()
        remove_result = self.remove_from_active(connection)
        self._idleConnections.append(connection)
        self._poolLock.release()

Here are the main parts of our implementation:

The constructor

Let’s have a look at the head of the class:

class ConnectionPool:
    _idleConnections: list[DbConnection] = None
    _activeConnections: list[DbConnection] = None
    _capacity: int = None
    _poolLock: threading.Lock = threading.Lock()

    def __init__(self, initial_connections: list[DbConnection]):
        self._idleConnections = initial_connections
        self._activeConnections = []
        self._capacity = len(initial_connections)

First of all we see that our class has four attributes:

1. A list of idle connections, or connections which can be used
2. A list of active connections, that is a list of connections in active use
3. A fixed capacity. In a more involved implementation, this could be dynamic
4. A poolLock which is used as we shall see to make the pattern thread-safe.

The get_connection() method

In the get_connection() method allows you to retrieve a connection from the pool if any. If no connection is available, it returns None

    def get_connection(self) -> DbConnection | None:
        self._poolLock.acquire()
        try:
            if len(self._idleConnections) == 0:
                return None
            result: DbConnection = self._idleConnections[0]
            self._idleConnections = self._idleConnections[1:]
            self._activeConnections.append(result)
            print(f"Gave out connection with Id {result.get_id()}")
            return result
        finally:
            self._poolLock.release()

A short summary:

1. We use the try-finally construction to make sure the lock is released. A finally block is always executed at the end of a try block
2. We check if we have connections, if none is available we return none.
3. If we have a connection available, we take the first available, remove it from our available connections, add it to the active connections and return it

The remove_from_active() method

This function is used to remove a connection from the list of active connections when you’re done with it and want to return it to the pool.

    def remove_from_active(self, connection: DbConnection) -> DbConnection | None:
        active_length = len(self._activeConnections)
        for index, conn in enumerate(self._activeConnections):
            if conn.get_id() == connection.get_id():
                self._activeConnections[active_length - 1], self._activeConnections[index] = self._activeConnections[
                    index], self._activeConnections[active_length - 1]
                self._activeConnections = self._activeConnections[:active_length - 1]
                return connection
        return None

This method might look cryptic so let’s go through it line by line:

1. We first establish the current number of activeconnections and store that in a variable.
2. Next we iterate over all the active connections. Note the use of the enumerate function, so we can automatically get an index too.
3. If we find the right connection, we swap the last element of the array with our current element. Then we chop off the last element.
4. And then we return the connection, to signal everything went well.
5. If no connection is found, we return None.

Also notice that we do not need the lock here, since this method is called from within a locked piece of code.

The return_connection() method

This method is used to return a connection to the pool when you no longer need it. It adds the connection back to the list of idle connections.

    def return_connection(self, connection: DbConnection):
        self._poolLock.acquire()
        remove_result = self.remove_from_active(connection)
        self._idleConnections.append(connection)
        self._poolLock.release()

Line by line:

1. We acquire our lock
2. Remove the connection from the active connections
3. Append the connection to our pool of idle connection
4. Release the lock

That wasn’t so hard.

Time to test

The testing section at the end demonstrates how to use the connection pool. We initialize the pool with two connections, get connections from the pool, return one, and then get another one to test the functionality.

if __name__ == "__main__":
    connections: list[DbConnection] = [DbConnection(1), DbConnection(2)]
    connection_pool = ConnectionPool(connections)

    first_connection = connection_pool.get_connection()
    print(f"Got connection with id: {first_connection.get_id()}")

    second_connection = connection_pool.get_connection()
    print(f"Got connection with id: {second_connection.get_id()}")

    third_connection = connection_pool.get_connection()
    if third_connection is None:
        print("Could not get third connection")

    connection_pool.return_connection(second_connection)

    third_connection = connection_pool.get_connection()
    print(third_connection.get_id())

Conclusion

The Object Pool pattern is a simple but powerful technique for managing and reusing resources efficiently. However, one drawback is that it doesn’t enforce returning objects to the pool, so you might want to consider mechanisms to ensure that in future implementations.