Skip to content

Use with connection pools

This guide shows how to test code that uses a connection pool with pytest-adbc-replay. The examples use adbc-poolhouse, but any pool library that calls adbc_clone() internally works the same way.

Configure auto-patching

List the driver module in adbc_auto_patch so the plugin intercepts connect() calls automatically:

[tool.pytest.ini_options]
adbc_auto_patch = [
    "adbc_driver_duckdb.dbapi",
]

[pytest]
adbc_auto_patch =
    adbc_driver_duckdb.dbapi

With auto-patching enabled, the connection returned by the driver's connect() is already a ReplayConnection. When the pool library calls adbc_clone() on it, each clone shares the same cassette path and configuration.

Create a pool fixture

Set up the pool in a session-scoped fixture and expose individual connections as function-scoped fixtures:

# conftest.py
import pytest
from adbc_poolhouse import DuckDBConfig, create_pool, close_pool


@pytest.fixture(scope="session")
def pool():
    p = create_pool(DuckDBConfig(database="/tmp/test.db"))
    yield p
    close_pool(p)


@pytest.fixture
def db_conn(pool):
    conn = pool.connect()
    yield conn
    conn.close()

pool.connect() calls adbc_clone() internally, so each checkout connection is a ReplayConnection sharing the same cassette.

Write a test

Use the db_conn fixture with @pytest.mark.adbc_cassette as usual:

import pytest


@pytest.mark.adbc_cassette("pool_query")
def test_pool_query(db_conn):
    with db_conn.cursor() as cur:
        cur.execute("SELECT 42 AS answer")
        result = cur.fetchall()
    assert result

Explicit wrap() approach

If you prefer not to use auto-patching, wrap the source connection manually before creating the pool:

# conftest.py
import pytest
import adbc_driver_duckdb.dbapi as duckdb
from adbc_poolhouse import DuckDBConfig, create_pool, close_pool


@pytest.fixture(scope="session")
def pool(adbc_replay):
    source = duckdb.connect(database="/tmp/test.db")
    wrapped = adbc_replay.wrap(source)
    p = create_pool(DuckDBConfig(database="/tmp/test.db"), source=wrapped)
    yield p
    close_pool(p)

The pool then calls adbc_clone() on the wrapped ReplayConnection, and all checkouts share the same cassette.

Warning

Cloned connections must be used sequentially. In typical single-threaded pytest runs, this is not a concern. Sequential access is only a limitation when testing multi-threaded server code with concurrent pool connections. See the explanation article for details on the failure mode.