Fix OperationalError: sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) SSL connection has been closed unexpectedly in FastAPI

This error occurs when the database connection in the session pool becomes stale or is closed by the server, but SQLAlchemy tries to reuse it. Fix it by configuring pool_pre_ping=True on your engine so SQLAlchemy tests each connection before using it, and ensure you properly close sessions after each request with a dependency that uses a finally block.

Reading the Stack Trace

Here's what each line means:

• File "/app/src/routes/users.py", line 15, in list_users: The route handler tried to query the database using a stale connection from the pool.
• File "/app/venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2348, in _handle_dbapi_exception: SQLAlchemy caught the database-level exception and is re-raising it as an OperationalError.
• File "/app/venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute: The underlying cursor.execute call failed because the connection was already closed by the database server.

Common Causes

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_db():
    db = SessionLocal()
    yield db
    db.close()  # Never reached if an exception is raised

db = SessionLocal()  # Global session, shared across all requests

@app.get("/users")
def list_users():
    return db.query(User).all()

The Fix

Enable pool_pre_ping=True so SQLAlchemy tests each connection with a lightweight ping before using it, automatically discarding stale connections. Set pool_recycle to proactively replace connections before the database server's idle timeout. Wrap the yield in try/finally to ensure sessions are always closed.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

def get_db():
    db = SessionLocal()
    yield db
    db.close()

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10,
    pool_recycle=300,
)
SessionLocal = sessionmaker(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Testing the Fix

import pytest
from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient
from app.main import app
from app.database import get_db


def test_db_session_closes_on_success():
    mock_session = MagicMock()

    def override_get_db():
        try:
            yield mock_session
        finally:
            mock_session.close()

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    response = client.get("/users")
    mock_session.close.assert_called_once()
    app.dependency_overrides.clear()


def test_db_session_closes_on_error():
    mock_session = MagicMock()
    mock_session.query.side_effect = Exception("DB error")

    def override_get_db():
        try:
            yield mock_session
        finally:
            mock_session.close()

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    response = client.get("/users")
    assert response.status_code == 500
    mock_session.close.assert_called_once()
    app.dependency_overrides.clear()


def test_list_users_returns_data():
    client = TestClient(app)
    response = client.get("/users")
    assert response.status_code == 200

Run your tests:

pytest tests/test_database.py -v

Pushing Through CI/CD

git checkout -b fix/fastapi-database-session,git add src/database.py tests/test_database.py,git commit -m "fix: add pool_pre_ping and proper session cleanup",git push origin fix/fastapi-database-session

Your CI config should look something like this:

name: CI
on:
  pull_request:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_DB: testdb
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: postgres
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          cache: 'pip'
      - run: pip install -r requirements.txt
      - run: pytest --tb=short -q
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/testdb

The Full Manual Process: 18 Steps

Here's every step you just went through to fix this one bug:

1. Notice the error alert or see it in your monitoring tool
2. Open the error dashboard and read the stack trace
3. Identify the file and line number from the stack trace
4. Open your IDE and navigate to the file
5. Read the surrounding code to understand context
6. Reproduce the error locally
7. Identify the root cause
8. Write the fix
9. Run the test suite locally
10. Fix any failing tests
11. Write new tests covering the edge case
12. Run the full test suite again
13. Create a new git branch
14. Commit and push your changes
15. Open a pull request
16. Wait for code review
17. Merge and deploy to production
18. Monitor production to confirm the error is resolved

Total time: 30-60 minutes. For one bug.

pip install bugstack

import bugstack

bugstack.init(api_key=os.environ["BUGSTACK_API_KEY"])

Deploying the Fix (Manual Path)

1. Run the test suite locally to confirm database sessions are properly managed.
2. Open a pull request with the pool configuration and session cleanup fix.
3. Wait for CI checks including database integration tests to pass.
4. Have a teammate review and approve the PR.
5. Merge to main and monitor database connection pool metrics in staging.

Frequently Asked Questions