connpool

A generic connection pool for Python 3.11+ — sync (threads) and asyncio. Standard library only (no third-party runtime dependencies). It pools anything: DB drivers, HTTP sessions, SSH clients, sockets… because the pool knows nothing about connections except what a small ConnectionFactory tells it.

from connpool import ConnectionPool, PoolConfig

pool = ConnectionPool.from_callables(make_conn, validate=is_alive, config=PoolConfig(max_size=20))
with pool:                                  # optional: warm-up + background sweeper
    with pool.connection() as conn:         # acquire -> use -> release (errors handled)
        conn.do_something()

Project layout

src/connpool/                 the library (imports: standard library only)
  core/            errors, clocks, PooledConnection/CloseReason, timeout helpers
  resilience/      breaker.py (circuit breaker), retry.py (backoff policy)
  policy/          config.py (PoolConfig), lifecycle.py (pure "when must a connection retire?" rules)
  factories/       sync_factory.py, async_factory.py  <- what YOU implement to pool anything
  observability/   events.py (PoolListener), stats.py (metrics)
  pools/           sync_pool.py, async_pool.py, registry.py (pool-per-key)
tests/             mirrors src/ (unittest, no dependencies)
  support/         fake factories and helpers
examples/          client code - NOT part of the library
  database/        generic DB-API factory (any driver), sqlite3 demo (runnable), psycopg demo
  http_client/  ssh/  async_usage/

Dependency direction is strictly downward (a layer only imports from layers listed above it): core → resilience → policy → factories/observability → pools. The top-level connpool/__init__.py is a facade re-exporting the public API, so from connpool import ConnectionPool keeps working even if internal modules move.

What you get

Writing a factory

class MyFactory(ConnectionFactory[MyConn]):
    def create(self):            return MyConn.connect(...)        # enforce your own connect timeout
    def validate(self, c):       return c.ping()                    # cheap liveness check
    def reset(self, c):          c.rollback()                       # scrub session state on release
    def close(self, c):          c.close()                          # default: calls c.close()
    def is_fatal_error(self, e): return isinstance(e, ConnectionError)

Async: subclass AsyncConnectionFactory, or pass sync/async callables to AsyncConnectionPool.from_callables. examples/database/dbapi_factory.py shows one factory that covers every PEP 249 database driver.

Configuration (PoolConfig)

Install, test, run examples

pip install -e .                                  # or: export PYTHONPATH=src
pip install -e ".[dev]"                           # + coverage/pytest/mypy
python -m unittest discover -s tests -t .         # 142 tests, ~2s
./run_tests.sh                                     # same, with a compact coverage report (gate: 100%)
mypy                                               # strict type check (config in pyproject.toml)
python examples/database/sqlite_demo.py

(pytest works too if installed; configuration is in pyproject.toml.) Examples importing paramiko or psycopg need those packages; the library never does.

Limits / extension points

• The breaker guards connection creation. To also trip on request failures wrap calls with CircuitBreaker.call().
• The breaker counts consecutive failed creations (one per logical acquire, retries included). A real backend failure or a create_timeout firing (the backend is too slow to connect — a genuine fault) both count; an acquire-timeout (caller impatience) or cancellation mid-create does not. A rate-based window is a drop-in replacement of one class.
• Leak detection is warn-only by design — the pool never yanks a connection a caller may still be using. It also only runs while the pool is open() (the maintenance sweep drives it), so registry pools need an explicit open(), and a warning can lag the real leak by up to one maintenance_interval.
• Not fork-safe. A pool open()ed before os.fork() (e.g. a pre-forking server like gunicorn with preload, or multiprocessing) shares live connections/file descriptors with the children — close the pool before forking and create a fresh one per process (max_size is per-process).
• Pool size is fixed at construction (PoolConfig is immutable); there is no runtime resize. Use a new pool, or a PoolRegistry keyed by profile, if you need different caps at runtime.
• Wake-up order of blocked threads is not strictly FIFO.
• Sync create() can't be interrupted by the pool; give your client a connect timeout.
• Listener hooks are synchronous; push heavy work to a queue.
• A factory's create() must return a distinct object each call; returning a shared/singleton connection is rejected with ConnectionCreateError (the pool tracks borrow state by object identity).
• acquire_wait percentiles are estimated over the last 1024 samples (bounded memory), not the full history.
• With validate_idle on, idle connections being pinged by the background sweep are briefly reserved and count as in_use (not idle) in stats() for the duration of that check — they are genuinely unavailable while validated, not borrowed by a caller.

Considered, not implemented

These are deliberate scope decisions, not oversights. For most, something already in the library covers the real need; adding them would make the pool heavier for every user to serve a minority. Listed here so the trade-off is explicit and the upgrade path is known.