如何传递sqlalchemy.exc.IntegrityError的异常详情？

我正在为网页爬取任务使用SQLAlchemy 2.0构建数据库：爬取网页后，结果暂存为project_min_example结构；爬取会话结束后，将数据持久化到数据库。为避免重复数据，将ID列设为唯一：

ID = Column(Integer, nullable=False, unique=True)

计划通过捕获sqlalchemy.exc.IntegrityError跳过已存在的数据。当前仅需处理ID唯一约束违反的异常（错误信息为(sqlite3.IntegrityError) UNIQUE constraint failed: projects.ID），其他完整性错误需抛出或通用处理。

我发现sqlalchemy.exc.IntegrityError可接收参数传递详情，但未找到正确用法。尝试以下代码未进入except分支：

# same imports as above
import sqlite3
# same code as above
    for project in project_min_example['projects_data']:
        print('*********************************************')
        print(f"Project {project['title']}, (ID {project['ID']}):")
        try:
            with Session(bind=engine) as session:
                project_SQLA = Project(project)
                session.add(project_SQLA)
                session.commit()
            print('successfully added to db')
        # except sqlalchemy.exc.IntegrityError as e:
        except sqlalchemy.exc.IntegrityError(statement='(sqlite3.IntegrityError) UNIQUE constraint failed: projects.Projekt_ID', params=project['ID'], orig=sqlite3.IntegrityError) as e:
            print(e)
            e.add_note('project already existing in database')
            print('project already existing')

现咨询：

1. 是否可通过sqlalchemy.exc.IntegrityError的参数传递错误详情？
2. 如何正确传递这些参数？

附最小实现示例代码：

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column
from sqlalchemy.sql.sqltypes import Integer, String
from sqlalchemy.orm import sessionmaker
import sqlalchemy.exc

base = declarative_base()   

class Project(base):
    
    __tablename__ = 'projects'       # Tabellenblatt

    index = Column(Integer, primary_key=True, autoincrement=True)   
    ID = Column(Integer, nullable=False, unique=True)
    title = Column(String(200), nullable=False)
    
    def __init__(self, project: dict) -> None:
        self.__dict__.update(project)       


if __name__ == '__main__':

    project_min_example = {
        'metadata': {'info_1': 'abc'},
        'projects_data': [
            {'ID': '1', 
             'title': 'project_1'},
            {'ID': '2', 
             'title': 'project_2'}
        ]
    }
    engine = create_engine("sqlite:///projects_minExample.db") 
    conn = engine.connect()
    Session = sessionmaker()    
    base.metadata.create_all(conn)
    for project in project_min_example['projects_data']:
        print('*********************************************')
        print(f"Project {project['title']}, (ID {project['ID']}):")
        try:
            with Session(bind=engine) as session:
                project_SQLA = Project(project)
                session.add(project_SQLA)
                session.commit()
            print('successfully added to db')
        except sqlalchemy.exc.IntegrityError as e:
            print(e)
            e.add_note('project already existing in database')
            print('project already existing')

问题1解答 #

sqlalchemy.exc.IntegrityError的参数是SQLAlchemy内部抛出异常时用来封装错误详情的，不能在except语句中通过传参数来匹配特定场景的异常——这种写法不符合Python异常捕获的逻辑，所以你的代码无法进入目标分支。

问题2解答 #

正确的做法是先捕获通用的IntegrityError，再在except分支内判断异常的具体内容，区分是否是ID唯一约束违反的情况。

修改后的代码示例 #

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column
from sqlalchemy.sql.sqltypes import Integer, String
from sqlalchemy.orm import sessionmaker
import sqlalchemy.exc
import sqlite3

base = declarative_base()   

class Project(base):
    
    __tablename__ = 'projects'

    index = Column(Integer, primary_key=True, autoincrement=True)   
    ID = Column(Integer, nullable=False, unique=True)
    title = Column(String(200), nullable=False)
    
    def __init__(self, project: dict) -> None:
        self.__dict__.update(project)       


if __name__ == '__main__':

    project_min_example = {
        'metadata': {'info_1': 'abc'},
        'projects_data': [
            {'ID': '1', 
             'title': 'project_1'},
            {'ID': '2', 
             'title': 'project_2'},
            # 新增重复ID测试数据
            {'ID': '1', 
             'title': 'project_1_duplicate'}
        ]
    }
    engine = create_engine("sqlite:///projects_minExample.db") 
    conn = engine.connect()
    Session = sessionmaker(bind=engine)    
    base.metadata.create_all(conn)
    for project in project_min_example['projects_data']:
        print('*********************************************')
        print(f"Project {project['title']}, (ID {project['ID']}):")
        try:
            with Session() as session:
                project_SQLA = Project(project)
                session.add(project_SQLA)
                session.commit()
            print('successfully added to db')
        except sqlalchemy.exc.IntegrityError as e:
            # 判断是否为ID唯一约束违反的异常
            if isinstance(e.orig, sqlite3.IntegrityError) and "UNIQUE constraint failed: projects.ID" in str(e.orig):
                e.add_note('project already existing in database')
                print('project already existing')
            else:
                # 其他完整性错误重新抛出
                raise e

关键说明 #

1. 异常判断逻辑：通过e.orig获取底层的sqlite3异常对象，先判断类型，再匹配错误信息字符串，确保只处理ID重复的场景。
2. 错误分流：非ID约束的完整性错误（比如非空字段为空）会被重新抛出，符合你“其他错误需抛出或通用处理”的需求。
3. Session优化：直接在sessionmaker绑定engine，无需在with Session时重复传参，代码更简洁。

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