Single sqlalchemy session in tests and asyncio tasks in FastApi. This session is in 'prepared' state; no further SQL can be emitted within this transaction

[YraganTron]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

# Database class
class Database:
    def __init__(self, *, db_url: str) -> None:
        self._engine = create_async_engine(db_url, echo=True)
        self._session_factory = async_scoped_session(
            sessionmaker(
                autocommit=False,
                autoflush=False,
                bind=self._engine,
                class_=AsyncSession,
                expire_on_commit=False,
            ),
            scopefunc=current_task,
        )

    @asynccontextmanager  # type: ignore
    async def session_maker(self) -> Callable[..., AbstractContextManager[AsyncSession]]:  # type: ignore
        session = self._session_factory()

        try:
            yield session
        except Exception:
            logger.exception('Session rollback because of exception')
            await session.rollback()
            raise
        finally:
            await self._session_factory.remove()

# Service class
class MyService:

    async def _record_processor(self, msg: ConsumerRecord) -> None:
        decode_msg = msg.value.decode()
        value = loads(decode_msg)
        data = self._mapping.get(msg.topic)

        scheme = self._get_kafka_model(model_name=data.dto).model_validate(value)

        async with self._db.session_maker() as session:
            await self._transit_events_dao.create_event(scheme, session)
            await self._create_or_update_point_statistic(scheme, session)
            await session.commit()

    async def _handle_kafka_records(self) -> None:
        async for record in self._kafka_consumer:
            try:
                await self._record_processor(record)
            except Exception as e:
                capture_exception(e)
                self._logger.error(e)

    def init_app(self, app: FastAPI) -> None:
        @app.on_event('startup')
        async def start() -> None:
            self._loop = get_running_loop()
            await self._kafka_consumer.start()
            self._kafka_consumer.subscribe(
                [
                    'my_topic'
                ]
            )
            self._consumer_task = self._loop.create_task(self._handle_kafka_records(), name='consumer')

        @app.on_event('shutdown')
        async def finish() -> None:
            if self._consumer_task is not None:
                self._consumer_task.cancel()
            await self._kafka_consumer.stop()

# My fixtures for tests. I have one session for my app and tests and I use 
# LifespanManager for run some code after startup my application(In my service deco 
# app.on_event('startup'))

from asgi_lifespan import LifespanManager
from httpx import AsyncClient

@pytest.fixture(scope='session', autouse=True)
def f_db(f_settings) -> Database:
    db = Database(db_url=f_settings.db_url, is_trace_active=False)
    return db


@pytest.fixture(scope='session', autouse=True)
async def _f_prepare_test_tables(*, f_db, event_loop, f_base_model) -> None:
    async with f_db._engine.begin() as conn:
        await conn.run_sync(f_base_model.metadata.drop_all)
        await conn.run_sync(f_base_model.metadata.create_all)


@pytest.fixture()
async def f_db_session(f_db) -> AsyncSession:
    async with f_db._engine.connect() as conn:
        trans = await conn.begin()

        await conn.begin_nested()

        async_session = AsyncSession(conn, expire_on_commit=False, autoflush=False, autocommit=False)

        @event.listens_for(async_session.sync_session, 'after_transaction_end')
        def end_savepoint(session, transaction) -> None:
            """here's an event.  inside the event we write blocking
            style code.  will this be fun to try to explain :)

            """

            if conn.closed:
                return

            if not conn.in_nested_transaction():
                conn.sync_connection.begin_nested()

        try:
            yield async_session
        except Exception:
            await async_session.rollback()
            raise
        finally:
            pass
            await async_session.close()
            await trans.rollback()


@pytest.fixture(autouse=True)
async def _f_set_session_mock(*, f_db, f_db_session) -> None:
    with pytest.MonkeyPatch.context() as m:
        m.setattr(f_db, 'session_maker', lambda: f_db_session)
        yield

    
@pytest.fixture(scope='session')
async def f_app(f_db):
    ApplicationContainer.events_db.override(f_db)
    app = create_web_app()
    return app
    
@pytest.fixture(scope='session', autouse=True)
async def f_app_client(*, f_app, f_last_api_prefix) -> AsyncClient:
    async with AsyncClient(
        app=f_app, base_url=f'http://test{f_last_api_prefix}', headers=PERMISSION_HEADERS
    ) as c, LifespanManager(f_app):
        REGISTRY.clear()
        await asyncio.sleep(0.5)
        yield c

# example test
@pytest.mark.parametrize('test_data', _get_merging_events_cases(), ids=lambda x: f"'{x.name}'")
async def test_merging_events(
    f_kafka_producer_integration_tests,
    f_db_session,
    test_data,
):

    for message in [message_1, message_2]:
        await send_kafka_msg(
            kafka_producer=f_kafka_producer_integration_tests, topic=topic_name, msg=json.dumps(message).encode()
        )

    # sleep here solved problem
    db_point_statistics = await wait_for_point_statistics_db_record(db_session=f_db_session, point_name=test_point_name)
    actual_event_db_point_statistics = db_point_statistics[0]
    assert actual_event_db_point_statistics.id is not None

Description

I have my fastapi app with sqlalchemy and tests for him. I can't attach minimal example code for reproduce it, but i understand problem. I'm attaching some code to help explain the problem.
I run all tests local sometime and I have error. And all my tests passed in CI always.

This session is in 'prepared' state; no further SQL can be emitted within this transaction

I use one session for test and for my app(This is convenient to keep the base clean). I have simple tests with send kafka message and check db. I use LifespanManager in my TestAsyncClient for run code after startup event. After that I create asyncio task and consume kafka messages and write to db. Since I run a asyncio task, I have a problem when a single session is used in different tasks.
I can add sleep before the request to the database in tests and this will solve the problem(this way there will never be a conflict within the same session). But it seems that the problem is fundamentally something else.

Maybe someone had similar problems and knows some kind of workaround?

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.104.1

Pydantic Version

2.6.1

Python Version

3.10.9

Additional Context

No response

[YuriiMotov]

I think you need to create less complicated code example to get the answer.. Try to get rid of unnecessary things and give us code we can run or at least understand easily.

[YuriiMotov]

Closing this. Feel free to reopen it whenever you are ready to provide less complicated code example (MRE)