db.py

async_engine = create_async_engine(
    cfg("DB_URL_ASYNC"), 
		pool_size=5, 
		echo=True, 
		max_overflow=10
)
TestingAsyncSessionLocal = sessionmaker(
    async_engine,
    expire_on_commit=False,
    autoflush=False,
    autocommit=False,
    class_=AsyncSession,
)

@pytest_asyncio.fixture(scope="function")
async def async_db_session():
    """The expectation with async_sessions is that the
    transactions be called on the connection object instead of the
    session object. 

    Detailed explanation of async transactional tests
    <https://github.com/sqlalchemy/sqlalchemy/issues/5811>
    """

    connection = await async_engine.connect()
    trans = await connection.begin()
    async_session = TestingAsyncSessionLocal(bind=connection)
    nested = await connection.begin_nested()

    @event.listens_for(async_session.sync_session, "after_transaction_end")
    def end_savepoint(session, transaction):
        nonlocal nested

        if not nested.is_active:
            nested = connection.sync_connection.begin_nested()

    yield async_session

    await trans.rollback()
    await async_session.close()
    await connection.close()