保存后的异步 SqlAlchemy 会话需要太多时间才能返回

Question

我正在从 SQLAlchemy 迁移到 SQLAlchemy[async]，使用 postgresql 作为我的数据库。我正在关注 SQL Alchemy 文档https://docs.sqlalchemy.org/en/14/_modules/examples/asyncio/async_orm.html

对代码进行所有更改后，我的测试变得非常慢

我添加了一堆日志以便打印丢失时间的位置。我也进行了分析，但 cProfile 没有显示任何相关内容。

仅在插入后我才在代码中出现这种延迟。在我分享的测试中，我先执行插入，然后执行检索，并且它仅在会话与操作一起使用后发生add。

engine = create_async_engine(DATABASE_URL,
                             pool_size=20,
                             max_overflow=2,
                             pool_recycle=300,
                             pool_pre_ping=True,
                             pool_use_lifo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)


@asynccontextmanager
async def session_scope() -> AsyncSession:
    """
    Context manager for API DB Sessions
    """
    try:
        async with async_session() as session:
            await session.begin()
            logger.info('Yielding session')
            yield session
            logger.info('Returned session')
            logger.info('Pre commit')
            await session.commit()
            logger.info('Commit finished')
    except Exception as exc:  # pylint: disable=W0703
        logger.exception('Exception on session')
        await session.rollback()
        raise exc 

这是我的客户

async def create_sync_job(entity: SyncJob) -> SyncJob:
    """
    Method for SyncJob creation
    :param sync_job:
    :return: SyncJob
    """
    start_time = time.time()
    async with session_scope() as session:
        logger.info(f'time spend in get session {time.time() - start_time}')
        start_time = time.time()
        session.add(entity)
        logger.info(f'time spend in saving {time.time() - start_time}')
        logger.info(f'Created sync job for user {entity.user_id} and {entity.pos_id}')
        return entity

还有我的测试

    async def test_create_sync_job(self):
        start_time = time.time()
        sync_job = sync_job_factory()
        start_time2 = time.time()
        created = await create_sync_job(sync_job)
        end_time = time.time()
        logger.info(f'spend time creating object {end_time - start_time2}')
        self.assertIsNotNone(created)
        end_time = time.time()
        logger.info(f'spend time {end_time-start_time}')
        created_db = await get_listing_sync_job_by_user(user_id=created.user_id)
        self.assertIsNotNone(created)

async with session.begin():我的测试效果很好，但是在返回控制之后之间有一个很大的时间窗口。这是在 messageReturned session和之间Pre commit。我有大量的测试要运行，这使得我的测试与非异步时相比非常慢。

{"@timestamp":"2022-02-17T15:08:34.496Z","log.level":"info","message":"Yielding session"}
{"@timestamp":"2022-02-17T15:08:34.496Z","log.level":"info","message":"time spend in get session 0.0013709068298339844"}
{"@timestamp":"2022-02-17T15:08:34.496Z","log.level":"info","message":"time spend in saving 2.288818359375e-05"}
{"@timestamp":"2022-02-17T15:08:34.496Z","log.level":"info","message":"Created sync job for user 61abcde4-0248-4c3f-a56e-4bdd4b6e20b0 and baea6703-87e3-4e5d-a318-5b67f4749c3c"}
{"@timestamp":"2022-02-17T15:08:34.496Z","log.level":"info","message":"Returned session"}
{"@timestamp":"2022-02-17T15:08:44.257Z","log.level":"info","message":"Pre commit"}
{"@timestamp":"2022-02-17T15:08:44.258Z","log.level":"info","message":"Commit finished"}
{"@timestamp":"2022-02-17T15:08:44.259Z","log.level":"info","message":"spend time creating object 9.763978958129883"}
{"@timestamp":"2022-02-17T15:08:44.259Z","log.level":"info","message":"spend time 9.768352031707764"}
{"@timestamp":"2022-02-17T15:08:44.259Z","log.level":"info","message":"Yielding session"}
{"@timestamp":"2022-02-17T15:08:44.274Z","log.level":"info","message":"Returned session"}
{"@timestamp":"2022-02-17T15:08:44.276Z","log.level":"info","message":"Pre commit"}
{"@timestamp":"2022-02-17T15:08:44.276Z","log.level":"info","message":"Commit finished"}

编辑：在代码中格式化

Answer 1

您是否尝试过执行会话而不是添加会话？我修复了数千条记录插入缓慢的问题。但我是在同步引擎上做到的。

或者替代方法是立即使用 session.commit 会话添加。

async def create_sync_job(entity: SyncJob) -> SyncJob:
"""
Method for SyncJob creation
:param sync_job:
:return: SyncJob
"""
start_time = time.time()
async with session_scope() as session:
    logger.info(f'time spend in get session {time.time() - start_time}')
    start_time = time.time()
    session.add(entity)
    session.commit() # <-
    logger.info(f'time spend in saving {time.time() - start_time}')
    logger.info(f'Created sync job for user {entity.user_id} and {entity.pos_id}')
    return entity