Object Pool

The Object Pool pattern allows for the reuse of expensive-to-create objects by keeping a pool of idle instances available for reuse.

This is useful in performance-sensitive or resource-constrained applications, such as:

  • Database connections

  • Network clients

  • Simulation/game objects

  • Pre-warmed service workers

pattern_kit provides two versions:

  • ObjectPool: For synchronous usage

  • AsyncObjectPool: For asyncio applications

Overview

When you acquire() an object, it is temporarily removed from the pool. When you’re done, you release() it back for future reuse.

If the pool is empty, a new object is created using the provided factory. The pool only limits how many idle objects it stores, not how many can be active at once.

Context manager support is available to automatically release objects:

with pool.borrow() as obj:
    obj.do_something()

async with async_pool.borrow() as obj:
    await obj.do_something()

Example Usage

from pattern_kit import ObjectPool

class Connection:
    def __init__(self):
        print("Opening connection")

pool = ObjectPool(factory=Connection, max_size=2)

conn1 = pool.acquire()
pool.release(conn1)

conn2 = pool.acquire()  # Reused connection
assert conn1 is conn2

Async Usage

from pattern_kit import AsyncObjectPool

class Worker:
    def __init__(self):
        self.jobs = []

pool = AsyncObjectPool(factory=Worker, max_size=5)

async def handle():
    async with pool.borrow() as worker:
        worker.jobs.append("job-123")

API Reference

class pattern_kit.creational.object_pool.ObjectPool(factory: Callable[[], T], max_size: int = 10)[source]

Bases: Generic[T]

A simple thread-safe object pool for synchronous use.

This pool reuses objects to avoid repeated construction. It does not strictly enforce a maximum number of active objects— max_size only limits how many can be stored for reuse.

acquire() T[source]

Acquire an object from the pool. If none are available, creates a new one.

borrow()[source]

Context manager version of acquire() + release().

clear() None[source]

Remove all idle objects from the pool.

release(obj: T) None[source]

Return an object to the pool for reuse. If the pool is full, the object is discarded.

class pattern_kit.creational.object_pool.AsyncObjectPool(factory: Callable[[], T], max_size: int = 10)[source]

Bases: Generic[T]

An asyncio-compatible object pool.

This pool reuses objects to avoid repeated construction. It does not strictly enforce a maximum number of active objects— max_size only limits how many can be stored for reuse.

async acquire() T[source]

Acquire an object from the async pool. Returns a new one if no reusable objects are available.

borrow()[source]

Async context manager version of acquire() + release().

async clear() None[source]

Remove all idle objects from the pool.

async release(obj: T) None[source]

Return an object to the pool. If the pool is full, the object is discarded.