具有异步调用的工厂男孩

Question

我正在尝试将工厂男孩与异步 SQLAlchemy、具体 SQLModel ORM 一起使用，并且在调用工厂男孩创建方法时遇到问题，它仅创建实例 od 对象，但从未存储在数据库中。我的工厂男孩是这样的

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


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

在 coftest 内部，我有 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())

我不确定会话是否有问题或异步调用是否有问题。当我对同一类型进行同步调用时，所有内容都已正确存储

Answer 1

只需重写_create工厂中的方法：

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

await比在异步测试或装置内部创建实例：

instance = await YourFactory()