# Stop opening new database connections in tests

> Tests that create an engine per test, use NullPool or close Django connections reconnect to the database over and over. Create the engine once per session and keep connections open.

Opening a database connection is far more work than running a simple query. The client opens a TCP socket (or a Unix socket), often negotiates TLS, authenticates, and the server sets up a session for it: PostgreSQL starts a new backend process for every connection. Then the connection is configured (time zone, encoding, isolation level) before the first query runs.

Applications avoid that cost with connection pools and persistent connections. Test suites often undo it by accident: a fixture creates a new engine for every test, a pool is disabled, or code under test closes the connection after each request. Each test then pays for one or more fresh connections.

## Where the reconnects come from

- **An engine per test.** A function-scoped fixture that calls `create_engine()` gets a new pool every time, so the first query of every test connects from scratch. Calling `engine.dispose()` in teardown closes all of them again.
- **`NullPool`.** `poolclass=NullPool` opens a new connection on every checkout and closes it on every check-in. It's common in async test setups, where it was added to work around "attached to a different loop" errors.
- **Explicit closes in Django.** `connection.close()`, `connections.close_all()` or `close_old_connections()` in a test, fixture or the code under test. Django reopens the connection on the next query.
- **New threads.** Django connections are per thread, so code that runs queries in a `ThreadPoolExecutor` or a background thread opens a connection in every new thread.
- **Raw driver calls.** A helper that calls `psycopg.connect()` for each query.

## SQLAlchemy: one engine per session

Create the engine once, in a session-scoped fixture, and keep the default `QueuePool`:

```diff
-@pytest.fixture
+@pytest.fixture(scope="session")
 def engine():
-    engine = create_engine(DATABASE_URL, poolclass=NullPool)
+    engine = create_engine(DATABASE_URL)
     yield engine
     engine.dispose()
```

Function-scoped fixtures then check a connection out of the pool and return it at the end of the test, which is cheap. If each test needs isolation, roll back a transaction on that connection rather than throwing it away (see [roll back per test with SQLAlchemy](https://flakytest.dev/guides/sqlalchemy-rollback-per-test/)).

If your app module creates its own engine at import time, make tests use it (or replace it once) instead of creating another engine per test.

### Async engines

`NullPool` in async tests usually hides a scope problem: a pooled asyncpg connection belongs to the event loop it was created on, and pytest-asyncio gives every test its own loop by default. Give the engine fixture and the event loop the same scope instead:

```python
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def engine():
    engine = create_async_engine("postgresql+asyncpg://postgres:postgres@localhost/test_myapp")
    yield engine
    await engine.dispose()
```

Tests that use it need to run in the session loop too (`@pytest.mark.asyncio(loop_scope="session")`, or the `asyncio_default_test_loop_scope` ini option). See [one event loop per async test](https://flakytest.dev/guides/pytest-asyncio-event-loop-scope/).

## Django: keep the connection open

Django's test runner and pytest-django keep one connection per database for the whole run, and Django's test client stops its requests from closing it. Inside a `TestCase` you can't reconnect at all: closing the connection in its transaction makes the next query fail with "Cannot open a new connection in an atomic block". So reconnects in Django suites come from transactional tests (`TransactionTestCase`, `@pytest.mark.django_db(transaction=True)`, `live_server`) and from fixtures outside a test's transaction:

- **Explicit closes.** `connection.close()` or `connections.close_all()` in a test, fixture or helper. Remove them from tests.
- **`close_old_connections()` in the code under test.** Worker loops and management commands often call it per unit of work. With the default `CONN_MAX_AGE = 0` every connection counts as obsolete, so each call closes it and the next query reconnects. In test settings, keep connections open:

```python
# settings/test.py
DATABASES["default"]["CONN_MAX_AGE"] = None  # persistent: not closed for its age
```

- **Threads.** Code that runs queries in a new thread opens a new connection in it. Run that work synchronously in tests (an executor you can swap, or a setting), which also lets it see the rows of a `TestCase` transaction.

## Keep the tests correct

- Pooled connections carry session state from one test to the next. SQLAlchemy's pool rolls back on check-in, but a committed `SET` or a temporary table stays with the connection. Avoid session-level state in tests, or reset it in the fixture that sets it.
- A session-scoped engine under pytest-xdist is per worker, so every worker still has its own connections and test database.
- Tests about reconnect behaviour (database restarts, `CONN_HEALTH_CHECKS`) should open connections on purpose. Mark them and keep them few.

## How flakytest finds it

flakytest looks for database samples inside connection setup: psycopg's `connect`, SQLAlchemy's pool creating a connection, and Django's `connect()`. When that adds up to at least 0.5 seconds and 2% of test time, it suggests **Tests open new database connections over and over**, with the tests and lines that triggered the connects. Run `flakytest compare` after the fix to see the time saved.

## Further reading

- [SQLAlchemy: connection pooling](https://docs.sqlalchemy.org/en/20/core/pooling.html)
- [Django: persistent connections](https://docs.djangoproject.com/en/stable/ref/databases/#persistent-connections)
- [pytest: fixture scopes](https://docs.pytest.org/en/stable/how-to/fixtures.html#scope-sharing-fixtures-across-classes-modules-packages-or-session)