如何将except: raise与SQLAlchemy异常集成？SQLAlchemyError无效

核心问题点 #

1. 会话传参错误：main.py中直接把async_session_maker传入retrieve_data_from_db，但函数要求的是AsyncSession实例，不是会话工厂。这会导致调用session.execute时根本没实际连接数据库，或触发的错误不在当前try块的捕获范围内。
2. 事务上下文与手动关闭冲突：async with my_async_session.begin()是SQLAlchemy的事务上下文管理器，它会自动处理事务的提交、回滚以及会话生命周期，手动调用await session.close()会干扰上下文正常工作，可能导致异常被吞掉。
3. 异常捕获时机问题：使用begin()的上下文管理器时，事务的提交/回滚操作在上下文退出时执行，如果查询本身没出错，但提交阶段出问题（比如约束冲突），异常会在async with块结束时抛出，而你的try块只包裹了execute步骤，根本捕获不到提交阶段的异常。

修正后的代码 #

db_commands.py #

如果需要手动控制事务与会话生命周期：

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

async def retrieve_data_from_db(query, my_async_session: AsyncSession):
    """连接数据库并返回给定查询的结果"""
    try:
        executed_query = await my_async_session.execute(query)
        await my_async_session.commit()
        return executed_query
    except SQLAlchemyError as e:
        await my_async_session.rollback()
        raise e
    finally:
        await my_async_session.close()

如果使用事务上下文管理器（推荐）：

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

async def retrieve_data_from_db(query, my_async_session: AsyncSession):
    """连接数据库并返回给定查询的结果"""
    async with my_async_session.begin():
        try:
            executed_query = await my_async_session.execute(query)
            return executed_query
        except SQLAlchemyError as e:
            raise e
    # 上下文结束后自动处理提交/回滚，无需手动close

main.py #

# 先创建会话实例，再传入函数
async with async_session_maker() as session:
    query = select(UserBase).where(UserBase.telegram_id == telegram_id)
    result = await retrieve_data_from_db(query, session)

额外说明 #

• 不需要导入所有SQLAlchemy异常，SQLAlchemyError是所有SQLAlchemy异常的基类，足够捕获相关错误。
• Spyder IDE本身不会导致异常吞掉的问题，核心是代码逻辑错误。
• 如果是只读查询，也可以考虑不开启事务，直接使用会话执行，减少不必要的事务管理开销。

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