Fix MissingGreenlet: sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here in FastAPI

This error occurs when you use SQLAlchemy's synchronous ORM operations inside an async FastAPI endpoint with an async engine. The sync API calls await_only() but no greenlet context exists. Fix it by using AsyncSession with async-compatible query methods like session.execute(select(...)) instead of session.query(...).

Reading the Stack Trace

Here's what each line means:

• File "/app/src/routes/products.py", line 18, in get_products: The async endpoint uses the synchronous db.query() API instead of the async session.execute() pattern.
• File "/app/venv/lib/python3.11/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 68, in await_only: SQLAlchemy's async engine requires greenlet context for sync-style calls, but the code is running outside that context.
• sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called: The async engine cannot perform synchronous IO. The query must use async-compatible methods.

Common Causes

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine(DATABASE_URL)

@app.get("/products")
async def get_products(db: AsyncSession = Depends(get_db)):
    products = db.query(Product).all()  # Sync API, won't work
    return products

@app.get("/orders/{order_id}")
async def get_order(order_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Order).where(Order.id == order_id))
    order = result.scalar_one()
    items = order.items  # Lazy load triggers sync IO

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)  # Wrong: sync sessionmaker with async engine

The Fix

Use async_sessionmaker with class_=AsyncSession for proper async database access. Replace the synchronous db.query() with await db.execute(select(...)) which is compatible with the async engine. Use result.scalars().all() to extract model instances from the result.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

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

@app.get("/products")
async def get_products(db: AsyncSession = Depends(get_db)):
    products = db.query(Product).all()
    return products

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

@app.get("/products")
async def get_products(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Product))
    products = result.scalars().all()
    return products

Testing the Fix

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app


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


@pytest.mark.asyncio
async def test_get_products_returns_correct_fields():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        response = await ac.get("/products")
    if response.json():
        product = response.json()[0]
        assert "id" in product
        assert "name" in product


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

Run your tests:

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

Pushing Through CI/CD

git checkout -b fix/fastapi-async-database,git add src/database.py src/routes/products.py tests/test_async_db.py,git commit -m "fix: use async_sessionmaker and select() for async database access",git push origin fix/fastapi-async-database

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+asyncpg://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 async test suite locally to confirm database queries work with the async engine.
2. Open a pull request with the async session and query migration.
3. Wait for CI checks including async database tests to pass.
4. Have a teammate review and approve the PR.
5. Merge to main and monitor query performance in staging.

Frequently Asked Questions