Factory boy with async calls

Question

I'm trying to use factory boy with async SQLAlchemy, concrete SQLModel ORM and having issue while calling factory boy create method it creates only instance od object but never stored in DB. My factory boy is looking like this

session = scoped_session(
    sessionmaker(class_=AsyncSession, bind=engine, autoflush=False)
)


class BaseFactory(SQLAlchemyModelFactory):
    class Meta:
        model = InitialModel
        abstract = True
        sqlalchemy_session = session

and inside coftest I have session that is needed for FastAPI

@pytest.fixture(scope="function")
async def session_fixture() -> AsyncSession:
    """Create a new session for a test and rollback changes after test"""
    async_session = sessionmaker(engine, class_=AsyncSession, autoflush=False)
    async with async_session() as session:
        yield session
        for table in reversed(SQLModel.metadata.sorted_tables):
            await session.execute(table.delete())

I'm not sure if there is problem with sessions or because of asynchronous calls. When I was working with synchronous calls on same type everything was stored propertly

Commit

Are you using pytest-asyncio and marking your test functions with @pytest.mark.asyncio? – Mimosa

tried with that but without success. it is running async as test but I think that problem is that factory create doesn't await on session.commit (if it provides it) and because od that it never store it in db – Commit

Answer 1

Just override _create method in your factory:

class YourFactory(SQLAlchemyModelFactory):
    class Meta:
        model = InitialModel
        abstract = True
        sqlalchemy_session = session

    @classmethod
    async def _create(cls, model_class, *args, **kwargs):
        instance = super()._create(model_class, *args, **kwargs)
        async with cls._meta.sqlalchemy_session as session:
            await session.commit()
        return instance

Than create instance with await inside your async tests or fixtures:

instance = await YourFactory()

Trodden

Answer 2

You can also use async-factory-boy, which is a factory_boy extension with asynchronous ORM support. Your factories can then use AsyncSQLAlchemyFactory (imported from async_factory_boy.factory.sqlalchemy) instead of SQLAlchemyModelFactory.

from async_factory_boy.factory.sqlalchemy import AsyncSQLAlchemyFactory

class TestModelFactory(AsyncSQLAlchemyFactory):
    class Meta:
        model = TestModel
        session = session

    name = Faker("name")
    created_at = Faker("date_time")

test_model = await TestModelFactory.create()

Chesterfield