transactional unit testing with pytest and sqlalchemy

Question

I am trying to implement transactional unit testing using ru test and sqlalchemy, but I get a "Runtime error task attached to a different loop". I think this is because pytest_asyncio creates another event loop in the fixture that returns the session, and directly in the test. Because of this, it is impossible to "await session.execute()" from the test. However, I have no idea how to fix it.

conftest.py

users = (
    UserSchema(
        phone_number="+71231231212",
        email="[email protected]",
        role="admin",
        discount=1,
        is_active=True,
    ),
    UserSchema(
        phone_number="+71234567891",
        email="[email protected]",
        role="GuestUser",
        discount=0,
        is_active=False,
    ),
)

# @pytest.fixture(scope="session", autouse=True)
# def event_loop():
#     try:
#         loop = asyncio.get_running_loop()
#     except RuntimeError:
#         loop = asyncio.new_event_loop()
#     yield loop
#     loop.close()


@pytest.fixture(scope="session")
async def async_engine():
    test_engine: AsyncEngine = create_async_engine(url=settings.TEST_DATABASE_URL)
    return test_engine


@pytest.fixture(autouse=True, scope="session")
async def prepare_database(async_engine):
    async with async_engine.begin() as conn:
        await conn.run_sync(BaseModel.metadata.drop_all)
        await conn.run_sync(BaseModel.metadata.create_all)
        for user in users:
            query = insert(UserModel).values(**user.model_dump())
            await conn.execute(query)


@pytest_asyncio.fixture(scope="function")
async def session(async_engine) -> None:
    conn = await async_engine.connect()
    trans = await conn.begin()
    async_session = AsyncSession(bind=conn, join_transaction_mode="create_savepoint")



    @event.listens_for(async_session.sync_session, "after_transaction_end")
    def restart_savepoint(session, transaction):
        if transaction.nested and not transaction._parent.nested:
            session.begin_nested()

    yield async_session

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

test_userdao.py

async def test_get_by_id(session: AsyncSession):
    assert 1 == 1
    query = select(UserModel).where(UserModel.id == 1)
    connection = await session.execute(text("SELECT 1"))
    result = await connection.execute(query)

I tried to create an event loop with my hands and make different implementations of the session fixture, tried this in the asynchronous version: task attached to a different loop