PyPool

A thread-safe Python connection pool with overflow management, configurable timeouts, and execution statistics.

Features

• Thread-Safe - Full locking for concurrent access
• Overflow Connections - Create temporary connections beyond pool size
• Configurable Timeouts - Pool-level and per-request timeout overrides
• Statistics - Track requests, overflows, timeouts, and returns
• Context Manager - Automatic cleanup with with statement
• Protocol-Based - Works with any object that has a close() method

Quick Start

import sqlite3
from pypool import ConnectionPool, PoolConfig

# Configure the pool
config = PoolConfig(
    pool_size=5,        # persistent connections
    max_overflow=10,    # extra connections when pool is full
    timeout=30.0,       # seconds to wait for a connection
)

pool = ConnectionPool(
    factory=lambda: sqlite3.connect("mydb.sqlite"),
    config=config,
)

# Use connections
conn = pool.acquire()
try:
    cursor = conn.cursor()
    cursor.execute("SELECT 1")
finally:
    pool.release(conn)

# Check statistics
print(pool.stats)
# {'total_requests': 1, 'total_created': 5, 'overflow_created': 0, ...}

Timeout Handling

from pypool import ConnectionPool, PoolConfig, PoolExhaustedError

pool = ConnectionPool(factory=create_conn, config=PoolConfig(pool_size=1, max_overflow=0, timeout=5))

try:
    conn = pool.acquire()
except PoolExhaustedError as e:
    print(f"Pool exhausted: size={e.pool_size}, waited={e.wait_time:.2f}s")

Running Tests

pytest