英文:
Factory boy with async calls
问题
我正在尝试在async SQLAlchemy、具体的SQLModel ORM中使用factory boy，并在调用factory boy的create方法时遇到问题，它只创建对象实例，但从未存储在数据库中。我的factory boy看起来像这样：
session = scoped_session(
sessionmaker(class_=AsyncSession, bind=engine, autoflush=False)
)
class BaseFactory(SQLAlchemyModelFactory):
class Meta:
model = InitialModel
abstract = True
sqlalchemy_session = session
在coftest中，我有一个用于FastAPI的session：
@pytest.fixture(scope="function")
async def session_fixture() -> AsyncSession:
"""Create a new session for a test and rollback changes after the 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 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
答案1
得分: 3
只需在你的工厂中覆盖 _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()
英文:
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()
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论