SQLAlchemy异步查询报错求助：with_only_columns参数传递问题

运行异步SQLAlchemy查询时出现错误：

The "entities" argument to Select.with_only_columns(), when referring to a sequence of items, is now passed as a series of positional elements, rather than as a list.

原始代码：

SessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,  # Используем асинхронный сеанс
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

async def get_claim_info(claim_id):
    async with SessionLocal() as session:
        try:
            stmt = select(Support.id, Support.Status, Support.Text).filter(Support.id == claim_id)
            result = await session.execute(stmt)

            claim_info = result.fetchone()

            if claim_info:
                claim_text = f"1: {claim_info[0]}\n2: {claim_info[1]}\n3: {claim_info[2]}"
                return claim_text
            else:
                return "none"
        except Exception as e:
            print("Error:", e)
            return "Error"
        finally:
            # Закрываем сессию после выполнения операции
            await session.close()

尝试过的修改（未解决）：

# 方式1
stmt = select(Support.id, Support.Status, Support.Text).filter(Support.id == claim_id).with_only_columns(Support.id, Support.Status, Support.Text)
# 方式2
stmt = select(Support).filter(Support.id == claim_id).with_only_columns(Support.id, Support.Status, Support.Text)
# 取值方式修改
claim_text = f"1: {claim_info.id}\n2: {claim_info.Status}\n3: {claim_info.Text}"

错误根源 #

错误提示指向with_only_columns()的传参规则：SQLAlchemy 2.0+要求该方法的列参数必须以位置参数形式传入（如with_only_columns(col1, col2)），不能传入列表（如with_only_columns([col1, col2])）。你贴的原始代码未调用with_only_columns()，推测是之前的代码版本误用列表传参导致报错。

修复后的代码 #

核心查询逻辑本身没问题，只需调整结果取值方式，并简化冗余操作：

SessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

async def get_claim_info(claim_id):
    async with SessionLocal() as session:
        try:
            # 直接指定查询列，无需额外调用with_only_columns
            stmt = select(Support.id, Support.Status, Support.Text).where(Support.id == claim_id)
            result = await session.execute(stmt)
            # 使用one_or_none()语义更明确：返回匹配行或None
            claim_info = result.one_or_none()
            
            if claim_info:
                # 方式1：索引取值（和原始逻辑兼容）
                claim_text = f"1: {claim_info[0]}\n2: {claim_info[1]}\n3: {claim_info[2]}"
                # 方式2：键名/属性名取值（需改用mappings()）
                # claim_info = result.mappings().one_or_none()
                # claim_text = f"1: {claim_info.id}\n2: {claim_info.Status}\n3: {claim_info.Text}"
                return claim_text
            else:
                return "none"
        except Exception as e:
            print("Error:", e)
            return "Error"

关键说明 #

1. 避免冗余操作：select()直接指定列时，不需要额外调用with_only_columns()。只有从全表查询（select(Support)）切换到指定列时，才需要使用该方法，且必须传位置参数，不能传列表。
2. 结果取值优化：
   • 用one_or_none()替代fetchone()，语义更清晰，明确返回单行或空值。
   • 如果想用属性名取值，需调用result.mappings().one_or_none()，返回的对象支持通过claim_info.id或claim_info['id']获取值。
3. 会话自动管理：async with SessionLocal()会在代码块结束时自动关闭会话，无需在finally中手动调用await session.close()。

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