如何在SQLAlchemy中调用异常?未找到行时的异常处理问询
嘿,我来帮你搞定SQLAlchemy里的异常调用和未找到行的处理问题~
1. 如何在SQLAlchemy中调用异常?
SQLAlchemy有一套专属的异常体系,所有核心异常类都在sqlalchemy.exc模块下,ORM相关的特殊异常则在sqlalchemy.orm.exc模块里。要调用异常,你只需要先导入对应的异常类,再用Python的raise关键字抛出即可。
举个基础例子,捕获数据库连接异常后,主动抛出带自定义信息的异常:
from sqlalchemy import create_engine from sqlalchemy.exc import SQLAlchemyError, OperationalError try: # 尝试连接一个不存在的数据库 engine = create_engine("sqlite:///nonexistent_database.db") conn = engine.connect() except OperationalError as original_err: print(f"连接数据库时出错: {original_err}") # 主动抛出更贴合业务场景的异常 raise SQLAlchemyError("无法初始化数据库连接,请检查配置") from original_err
常用的异常类包括:
SQLAlchemyError:所有SQLAlchemy异常的基类,可用于捕获通用异常OperationalError:数据库操作相关错误(比如连接失败、执行无效SQL)IntegrityError:违反数据库完整性约束(比如唯一键重复)
2. 针对未找到行的情况,该如何调用异常?
如果是用SQLAlchemy ORM查询,未找到行最常用的异常是NoResultFound(还有对应的MultipleResultsFound,当查询返回多行但你期望仅一行时抛出),这两个异常都在sqlalchemy.orm.exc模块下。
场景1:使用one()方法查询
当你用query.one()期望返回恰好一行结果时,如果没有找到匹配行,SQLAlchemy会自动抛出NoResultFound,你可以捕获处理,也可以主动抛出给上层:
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound from your_models import User # 替换成你的实际模型类 Session = sessionmaker(bind=your_engine) session = Session() try: # 查询ID为999的用户,期望唯一结果 user = session.query(User).filter(User.id == 999).one() except NoResultFound: print("没有找到ID为999的用户") # 主动抛出异常给上层业务逻辑处理 raise NoResultFound(f"用户ID {999}不存在,请检查输入") except MultipleResultsFound: print("找到多个ID为999的用户,数据可能存在重复")
场景2:使用first()方法查询
如果用query.first(),它不会自动抛出异常,无结果时会返回None,这时候你需要手动判断并抛出异常:
user = session.query(User).filter(User.id == 999).first() if not user: # 主动抛出未找到的异常 raise NoResultFound(f"未找到ID为999的用户")
场景3:使用SQLAlchemy Core(原生SQL查询)
如果是用核心API执行原生SQL,查询结果返回None时也需要手动判断并抛出:
from sqlalchemy.exc import NoResultFound result = your_engine.execute("SELECT * FROM users WHERE id = 999").fetchone() if not result: raise NoResultFound("未找到符合条件的行")
如果之前你的尝试没成功,大概率是没导入正确的异常类(比如把NoResultFound导入错了模块),或者用了不会自动抛出异常的查询方法(比如first())却没手动判断结果~
内容的提问来源于stack exchange,提问作者user9377278
相关产品推荐
相关产品推荐

