Fix RuntimeError: RuntimeError: Event loop is closed in FastAPI

This error occurs when an async startup event or lifespan handler tries to use the event loop after it has been closed, typically because the async resource initialization fails and cleanup runs against a closed loop. Fix it by using FastAPI's lifespan context manager pattern to properly initialize and clean up async resources like database pools and HTTP clients.

Reading the Stack Trace

Here's what each line means:

• File "/app/src/main.py", line 12, in startup_event: The startup event handler attempts to create an async database pool, which may fail if the database is unreachable.
• File "/app/venv/lib/python3.11/site-packages/asyncpg/pool.py", line 401, in _async__init__: asyncpg tries to initialize the connection pool, and failure during startup leaves the app in a broken state.
• File "/app/src/main.py", line 22, in shutdown_event: The shutdown handler tries to close the pool on a closed event loop because the startup failed and the loop was already torn down.

Common Causes

from fastapi import FastAPI
import asyncpg

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    app.state.db_pool = await asyncpg.create_pool(DATABASE_URL)

@app.on_event("shutdown")
async def shutdown_event():
    await app.state.db_pool.close()  # Fails if startup_event failed

@app.on_event("startup")
async def startup():
    app.state.redis = await aioredis.from_url(REDIS_URL)
    app.state.db = await asyncpg.create_pool(DATABASE_URL)
    # If Redis connects but DB fails, Redis connection leaks

@app.on_event("shutdown")
async def shutdown():
    await app.state.redis.close()  # AttributeError if redis was never set
    await app.state.db_pool.close()

The Fix

Use the lifespan context manager pattern instead of deprecated on_event decorators. The code before yield runs on startup, and the code after yield runs on shutdown. This pattern guarantees cleanup runs in the same event loop context and only executes if startup succeeded.

from fastapi import FastAPI
import asyncpg

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    app.state.db_pool = await asyncpg.create_pool(DATABASE_URL)

@app.on_event("shutdown")
async def shutdown_event():
    await app.state.db_pool.close()

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg
import logging

logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    logger.info("Connecting to database...")
    app.state.db_pool = await asyncpg.create_pool(DATABASE_URL)
    logger.info("Database pool created.")
    yield
    # Shutdown
    logger.info("Closing database pool...")
    await app.state.db_pool.close()
    logger.info("Database pool closed.")

app = FastAPI(lifespan=lifespan)

Testing the Fix

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from unittest.mock import patch, AsyncMock
from app.main import app


@pytest.mark.asyncio
async def test_app_starts_and_serves_requests():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        response = await ac.get("/health")
    assert response.status_code == 200


@pytest.mark.asyncio
async def test_db_pool_is_available_after_startup():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        response = await ac.get("/health")
        assert response.status_code == 200


@pytest.mark.asyncio
async def test_app_handles_multiple_requests():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        for _ in range(10):
            response = await ac.get("/health")
            assert response.status_code == 200

Run your tests:

pytest tests/test_startup.py -v --asyncio-mode=auto

Pushing Through CI/CD

git checkout -b fix/fastapi-startup-lifespan,git add src/main.py tests/test_startup.py,git commit -m "fix: migrate from on_event to lifespan context manager for proper startup/shutdown",git push origin fix/fastapi-startup-lifespan

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 --asyncio-mode=auto
        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 the app starts and shuts down cleanly.
2. Open a pull request with the lifespan migration.
3. Wait for CI checks to pass on the PR.
4. Have a teammate review and approve the PR.
5. Merge to main and verify the app starts correctly in staging with real database connections.

Frequently Asked Questions