SQLAlchemy调用MySQL存储过程获取多结果集报错求助

问题描述 #

使用Python的SQLAlchemy异步调用MySQL存储过程，期望获取两个查询结果集，但执行时报错：

Error creating user: This result object does not return rows. It has been closed automatically.

直接在MySQL客户端执行存储过程可正常创建数据并返回结果。

Python异步函数代码 #

async def create_user(db: AsyncSession, user: UserInfo):
    try:
        result_proxy = await db.execute(
            text("CALL create_user(:account)"), 
            {
                "account": user.account
            }
        )
        logging.info("result_proxy: %s", result_proxy)
        # Fetch the first result set
        db_user = result_proxy.fetchone()

        logging.info("db_user: %s", db_user)
        
        # Move to the next result set
        if result_proxy.nextset():
            # Fetch the second result set
            db_user_setting = result_proxy.fetchone()
            logging.info("db_user_setting: %s", db_user_setting)
        else:
            db_user_setting = None

    except SQLAlchemyError as e:
        logging.error("Error creating user: %s", e)
        await db.rollback()
        raise HTTPException(status_code=500, detail="Database creating user failed.")

    return db_user, db_user_setting

MySQL存储过程代码 #

CREATE DEFINER=`test`@`%` PROCEDURE `data`.`create_user`(
    IN p_account VARCHAR(255)
)
BEGIN
    DECLARE v_user_id VARCHAR(255);
 
    DECLARE EXIT HANDLER FOR SQLEXCEPTION
    BEGIN
        -- Rollback the transaction in case of an error
        ROLLBACK;
    END;

    SET v_user_id = UUID_SHORT();
       
    START TRANSACTION;
    
    INSERT INTO data.user_info (
        account,
        user_id
    )
    VALUES(
        p_account,
        v_user_id
    );

    INSERT INTO data.user_setting (
        user_id,
        created_name
    )
    VALUES(
        v_user_id,
        'Javis'
    );
    
    COMMIT;
   
    -- Select the newly created records to return them as a result set
    SELECT * FROM data.user_info WHERE user_id = v_user_id;
    SELECT * FROM data.user_setting  WHERE user_id = v_user_id;
   
END

排查建议 #

1. 跳过存储过程中更新操作的结果集 #

MySQL存储过程中执行INSERT等DML操作时，会返回受影响行数作为一个结果集，你的存储过程包含2个INSERT，所以在两个SELECT结果集之前，还有2个计数结果集。直接读取第一个结果集拿到的是INSERT的行数，而非查询数据，此时调用fetchone()会因结果集无行数据报错。

修改代码，先跳过这些计数结果集，再读取目标SELECT结果：

async def create_user(db: AsyncSession, user: UserInfo):
    try:
        result_proxy = await db.execute(
            text("CALL create_user(:account)"), 
            {"account": user.account},
            stream_results=True  # 关键参数：保持结果集不自动关闭
        )
        # 跳过2个INSERT的结果集
        for _ in range(2):
            result_proxy.nextset()
        
        # 读取第一个SELECT结果
        db_user = result_proxy.fetchone()
        logging.info("db_user: %s", db_user)
        
        # 切换到第二个SELECT结果集并读取
        if result_proxy.nextset():
            db_user_setting = result_proxy.fetchone()
            logging.info("db_user_setting: %s", db_user_setting)
        else:
            db_user_setting = None

    except SQLAlchemyError as e:
        logging.error("Error creating user: %s", e)
        await db.rollback()
        raise HTTPException(status_code=500, detail="Database creating user failed.")

    return db_user, db_user_setting

2. 添加stream_results=True参数 #

异步模式下，SQLAlchemy默认会在执行后自动关闭结果对象，添加stream_results=True可以让结果集保持打开状态，直到你完成所有结果集的读取操作，避免出现结果对象已关闭的错误。

3. 避免事务嵌套冲突 #

存储过程内部已经包含START TRANSACTION和COMMIT，而SQLAlchemy的AsyncSession默认会开启事务，可能导致事务嵌套，影响结果集的返回。可以尝试在调用存储过程时临时启用自动提交：

async def create_user(db: AsyncSession, user: UserInfo):
    try:
        # 临时切换到autocommit模式执行存储过程
        async with db.begin():
            db.autocommit = True
            result_proxy = await db.execute(
                text("CALL create_user(:account)"), 
                {"account": user.account},
                stream_results=True
            )
            db.autocommit = False
        
        # 跳过2个INSERT的结果集
        for _ in range(2):
            result_proxy.nextset()
        
        # 读取目标结果集逻辑...

    except SQLAlchemyError as e:
        logging.error("Error creating user: %s", e)
        await db.rollback()
        raise HTTPException(status_code=500, detail="Database creating user failed.")

    return db_user, db_user_setting

4. 检查MySQL驱动版本 #

确保使用的异步MySQL驱动（如aiomysql）版本是最新的，旧版本可能存在存储过程多结果集处理的兼容性问题。

内容的提问来源于stack exchange，提问作者Youshikyou