pythonexceptionsqlalchemy

sqlalchemy.exc.IntegrityError: How to pass exception details?


For a web scraping task I am building up a database using SQLA 2.0:

A weppage is scraped, the results are temporarily kept according to `project_min_example ` (struture in code block below).
After a scraping session is finsihed I want to permanently store them to a database using SQLAlchemy. To avoid having the same data multiple times in my database due to finding the same data during several scraping session the column ID is configured to only accept unique values:

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

The idea is to skip adding data which is already present in the database by catching `sqlalchemy.exc.IntegrityError`.

Minimum example:

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')

Right now I only want to handle exactly the exception case as described above which produces this error message:
(sqlite3.IntegrityError) UNIQUE constraint failed: projects.ID
[SQL: INSERT INTO projects ("ID", title) VALUES (?, ?)]
[parameters: ('1', 'project_1')]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
(I had to format the error message as code in order to enable posting my question)

Other possible integrity error types shall really produce errors, or be handled in a more general way (not implemented in the code example).
I found out, that `sqlalchemy.exc.IntegrityError` accepts arguments and I believe they could be used to pass error details. Unfortunately the documentation I found did not help me to understand how to correctly pass them.

My questions:
- Can I use the `sqlalchemy.exc.IntegrityError` arguments to provide error details ?
- How to properly pass the arguments ?

I tried:

# 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')

--> the `except` block won't be entered


Solution

  • Finally I came to this solution, using
    the expected errors properties:

    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:
        if e.orig.sqlite_errorname == 'SQLITE_CONSTRAINT_UNIQUE' and e.orig.args[0] == 'UNIQUE constraint failed: projects.Projekt_ID':
            print('project already existing')
        else: 
            print('unexpected error (classs sqlalchemy.exc.IntegrityError)')