claude-memory/graph/solutions/pytest-asyncio-sqlalchemy-async-db-test-fixture-pattern-fecad3.md
Cal Corum b140d4d82a migrate: 313 memories from MemoryGraph
- 313 new markdown files created
- 30 relationships embedded
- 313 entries indexed
- State initialized with usage data
2026-02-13 11:11:48 -06:00

1.9 KiB

id type title tags importance confidence created updated
fecad34a-4cc1-4f00-878c-393a9ecb5df0 solution pytest-asyncio SQLAlchemy async DB test fixture pattern
mantimon-tcg
pytest
pytest-asyncio
sqlalchemy
async
testing
pattern
0.8 0.8 2026-01-27T15:50:36.216118+00:00 2026-01-27T15:50:36.216118+00:00

When testing SQLAlchemy async with pytest-asyncio, the fixture teardown runs in a DIFFERENT event loop than the test body, causing 'Future attached to different loop' errors on async cleanup.

SOLUTION:

  1. Use sync psycopg2 for fixture setup/teardown operations (migrations, truncate)
  2. Create fresh NullPool engine per test (no connection reuse)
  3. Use TRUNCATE via sync psycopg2 in teardown instead of async rollback
  4. Suppress warnings for connection cleanup (harmless GC noise)

KEY CODE (conftest.py):

def truncate_all_tables():
    conn = psycopg2.connect(**DB_PARAMS)
    try:
        conn.autocommit = True
        with conn.cursor() as cur:
            for table in TABLES_TO_TRUNCATE:
                cur.execute(f'TRUNCATE TABLE {table} CASCADE')
    finally:
        conn.close()

@pytest_asyncio.fixture
async def db_session():
    engine = create_async_engine(URL, poolclass=pool.NullPool)
    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    session = session_factory()
    try:
        yield session
    finally:
        try:
            await session.close()
        except RuntimeError:
            pass  # Ignore event loop errors
        truncate_all_tables()  # SYNC cleanup always works

ADDITIONAL NOTES:

  • When testing relationship loading (selectinload), must expire() the parent object first if children were added after initial load
  • ON DELETE SET NULL requires passive_deletes=True on relationship to let DB handle it
  • Suppress warnings in pyproject.toml: 'ignore:The garbage collector is trying to clean up:sqlalchemy.exc.SAWarning'