Deal with Common Types of SQLAlchemy Exceptions for Running SQL Queries in Python

When working with databases using SQLAlchemy, many developers may simply catch the base exception SQLAlchemyError, or worse, the base Python Exception. This is not a good practice because it makes the code work like a black box and we cannot anticipate what types of exceptions can be raised.

In this post, we will introduce some most commonly encountered SQLAlchemy exceptions with simple and reproducible examples. Having a knowledge of these specific exceptions and using them properly in our code can make our code more robust and more professional.

---

OperationalError

The first exception type we are going to introduce is OperationalError, which is a very common one and normally occurs when it fails to connect to the database due to various reasons. We can simply provide an incorrect password and trigger the error:

import logging

from sqlalchemy import create_engine, text
from sqlalchemy.exc import OperationalError

try:
    db_url = "mysql+pymysql://root:wrong-password@localhost:13306/data"
    engine = create_engine(db_url, pool_size=5, pool_recycle=3600)
    conn = engine.connect()
except OperationalError as err:
    logging.error("Cannot connect to DB %s", err)
    raise err

# OperationalError: (pymysql.err.OperationalError) (1045, "Access denied for user ''root''@''172.17.0.1'' (using password: YES)")

Actually, not just incorrect credentials, any issue that can cause DB connection failure will raise OperationalError, including

• Incorrect host or port.
• Incorrect username or password.
• Incorrect default database.
• The database cannot be reached. It can just be down or not reachable through the network.

Besides DB connection issues, lock and deadlock will also raise OperationalError, let’s try to trigger one in our database.

In the MySQL console, workbench, or DBeaver, run the following queries to lock the student_scores table:

START TRANSACTION;
SELECT * FROM data.student_scores FOR UPDATE;

Then in Python, run the following code:

import logging

from sqlalchemy import create_engine, text
from sqlalchemy.exc import OperationalError

try:
    db_url = "mysql+pymysql://root:root@localhost:13306/data"
    engine = create_engine(db_url, pool_size=5, pool_recycle=3600)
    conn = engine.connect()
except OperationalError as err:
    logging.error("Cannot connect to DB %s", err)

try:
    conn.execute(text("UPDATE data.student_scores SET score = 100 WHERE student_id = 1"))
except OperationalError as err:
    logging.error("Query failed due to lock/dead lock issues: %s", err)
finally:
    conn.close()

# OperationalError: (pymysql.err.OperationalError) (1205, ''Lock wait timeout exceeded; try restarting transaction'')

As demonstrated, OperationalError is raised when the query failed due to table-locking issues. It’s more difficult to trigger a deadlock, but the error raised will also be OperationalError when it does happen.

We should close the connection in the finally block so it can always be closed no matter whether an exception is raised or not.

Besides, don’t forget to run ROLLBACK in the MySQL console, workbench, or DBeaver where the table is locked, so the table can be unlocked.

---

IntegrityError

Another common SQLAlchemy exception is IntegrityError, which is raised when an integrity constraint is violated in a database. The integrity constraint can be the primary key, unique key, or foreign key.

Since the student_scores table has a primary key on the student_id and subject columns, we can trigger an IntegrityError by inserting a row with existing subject_id and subject:

try:
    conn.execute(
        text(
            """
            INSERT INTO `data`.student_scores
                (student_id, subject, score)
            VALUES
                (1, ''Math'', 100)
            """
        )
    )
    conn.commit()
except IntegrityError as err:
    logging.error("Duplicate entry: %s", err)
finally:
    conn.close()

# IntegrityError: (pymysql.err.IntegrityError) (1062, "Duplicate entry ''1-Math'' for key ''student_scores.PRIMARY''")

For primary key and unique key constraints, depending on your actual use case, you may want to simply ignore the IntegrityError error or update existing records when such an error occurs. However, for foreign key constraints, normally the error cannot be handled automatically and you need to fix the foreign key issue manually.

---

DataError

DataError is another very commonly encountered SQLAlchemy exception. It is raised when the data cannot be processed by SQLAlchemy. For example, when the data type cannot be converted automatically. Let’s trigger a DataError by updating the score to be a string value that cannot be converted to a number:

try:
    conn.execute(
        text(
            """
            UPDATE data.student_scores
            SET score = ''A''
            WHERE student_id = 1
            """
        )
    )
    conn.commit()
except DataError as err:
    logging.error("Incorrect data: %s", err)
finally:
    conn.close()

# DataError: (pymysql.err.DataError) (1366, "Incorrect integer value: ''A'' for column ''score'' at row 1")

Since the score column is an integer type in our DB, the string “A” cannot be saved. However, the string “100” can be saved because it can be converted to a number.

Besides data type conversion issues, DataError can also be raised when the string value provided exceeds the predefined maximum length for a VARCHAR field:

try:
    conn.execute(
        text(
            """
            INSERT INTO `data`.student_scores
                (student_id, subject, score)
            VALUES
                (4, ''MathMathMathMathMathMathMathMathMathMathMathMathMath'', 100)
            """
        )
    )
    conn.commit()
except DataError as err:
    logging.error("Incorrect data: %s", err)
finally:
    conn.close()

# DataError: (pymysql.err.DataError) (1406, "Data too long for column ''subject'' at row 1")

DataError is raised because the subject column has a maximum length of 50 characters.

---

SQLAlchemyError

SQLAlchemyError is the base class for all exceptions raised by SQLAlchemy:

from sqlalchemy.exc import DataError, IntegrityError, OperationalError, SQLAlchemyError

issubclass(DataError, SQLAlchemyError)
# True
issubclass(IntegrityError, SQLAlchemyError)
# True
issubclass(OperationalError, SQLAlchemyError)
# True

In general, it’s a good practice to catch specific exceptions like DataError, IntegrityError, OperationalError rather than the base exception SQLAlchemyError, because it lets you know your code better and allows you to handle different types of errors in different ways. Nonetheless, SQLAlchemyError can be used as the fallback exception when no specific exception can be caught. For example:

try:
    conn.execute(text("SOME QUERY..."))
    conn.commit()
except DataError as err:
    logging.error("Incorrect data: %s", err)
except IntegrityError as err:
    logging.error("Duplicate entry: %s", err)
except OperationalError as err:
    logging.error("Query failed due to lock/dead lock issues: %s", err)
except SQLAlchemyError as err:
    logging.error("Unexpected SQLAlchemy error: %s", err)
finally:
    conn.close()

For a specific query, you don’t need to catch all the listed exceptions as demonstrated above and can just choose the potential ones depending on your query.

To avoid code repetition, we can put this code block in a library file so we don’t need to repeat the exception-catching code whenever a query is executed.

---

In this post, we have introduced some most commonly encountered SQLAlchemy exceptions. Having a knowledge of these specific exceptions and using them properly in your code can make your code more robust and more professional. Now you can improve your code and stop simply catching SQLAlchemyError or Exception in your code.