how to catch specific pyodbc error message

Question

I trid the following code,

import pyodbc
try:
    pyodbc.connect('DRIVER={%s};SERVER=%s;DATABASE=%s;UID=%s;PWD=%s' % (driver, server, database, uid, password))
except pyodbc.Error, err:
    logging.warn(err)

The error message format i get is

('HY000', "[HY000] [MySQL][ODBC 5.1 Driver]Access denied for user 'root'@'192.168.2.27' (using password: YES) (1045) (SQLDriverConnect)")

I want to receive just the message part of the error i.e.

Access denied for user 'root'@'192.168.2.27'(using password: YES)

I dont know if I can catch errors specifically like, driver not found, host down etc..

I also tried catching errors as:

 except pyodbc.OperationalError, err:
    logging.warn(err)
except pyodbc.DataError, err:
    logging.warn(err)
except pyodbc.IntegrityError, err:
    logging.warn(err)
except pyodbc.ProgrammingError, err:
    logging.warn(err)
except pyodbc.NotSupportedError, err:
    logging.warn(err)
except pyodbc.DatabaseError, err:
    logging.warn(err)
except pyodbc.Error, err:
    logging.warn(err)

but the last one always catches the error.

Fruthermore i saw the pyodbc.Error.message is always empty. How can i get just the message in the error.

Thanks

Zoophobia

Answer 1

pyodbc seems to just wrap the errors/exceptions from the underlying ODBC implementation, so it's unlikely that you will be able to do this.

Boyles

Answer 2

This worked for me.

    try:
        cnxn = pyodbc.connect(...)
    except pyodbc.Error as ex:
        sqlstate = ex.args[0]
        if sqlstate == '28000':
            print("LDAP Connection failed: check password")

There are different SQLSTATES and you can have if-else statements to print out the cause.

Similarly,

  try:
        cnxn = pyodbc.connect(...)
  except pyodbc.Error as ex:
        sqlstate = ex.args[1]
        print(sqlstate) 

will give you the second part of the error with description. For exampleex.args[0] give you 28000 and ex.args[1] gives [28000] LDAP authentication failed for user 'user' (24) (SQLDriverConnect)

You can then use String manipulation techniques there to just print out what you want. Hope this helps.

Iata

This also works with pypyodbc, when you replace pyodbc with the correct library name. – Disbursement

Answer 3

It's been very long since op asked this question, but here goes a snippet of code to parse out pyodbc error messages into nice Python exceptions that can be used. This is also meant to be extended, I didn't handle every possible sqlserver error code.

import re
from enum import Enum, IntEnum, unique


class PyODBCError(Exception):
    """
    Handle errors for PyODBC. Offers a error message parser
    to apply specific logic depending on the error raise

    ODBC error identifier: 23000

    pyodbc_error_message (str) -- message raised by PyODBC
        Example:
            [23000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server] \
            Cannot insert explicit value for identity column in table \
            'building' when IDENTITY_INSERT is set to OFF.
            (544) (SQLExecDirectW) \
    """

    error_pattern = re.compile(
        r"\[(?P<error_id>.*?)\] \[(?P<operator>.*?)\]\[(?P<driver>.*?)\]\[(?P<database_type>.*?)\](?P<error_message>.+?(?= \()) \((?P<sql_server_error_id>\d*?)\) \(SQLExecDirectW\)"
    )
    sql_error_code_pattern = re.compile(r"\((?P<sql_server_error_code>\d*?)\) \(SQLExecDirectW\)")
    column_pattern = re.compile(r"column \'(?P<column_name>.+?)\'")
    table_pattern = re.compile(r"table \'(?P<table_name>.+?)\'")
    pyodbc_error_code = 'HY000'

    def __init__(self, pyodbc_error_message: str) -> None:
        self._parse_error_message(pyodbc_error_message)

    def __str__(self) -> str:
        return self.error_message

    def _parse_error_message(self, pyodbc_error_message: str) -> None:
        m = re.match(self.error_pattern, pyodbc_error_message)
        self.operator = m.group('operator')
        self.error_id = m.group('error_id')
        self.driver = m.group('driver')
        self.database_type = m.group('database_type')
        self.error_message = m.group('error_message')
        self.sql_server_error_id = m.group('sql_server_error_id')

    @classmethod
    def get_message(cls, pyodbc_exception: Exception) -> str:
        if pyodbc_exception.args[1] == cls.pyodbc_error_code:
            return pyodbc_exception.args[0]
        else:
            return pyodbc_exception.args[1]

    @classmethod
    def get_pyodbc_code(cls, pyodbc_exception: Exception) -> str:
        if pyodbc_exception.args[1] == cls.pyodbc_error_code:
            return pyodbc_exception.args[1]
        else:
            return pyodbc_exception.args[0]

    @staticmethod
    def get_exception(error_code: int):
        return {
            515: IdentityInsertNull,
            544: IdentityInsertSetToOff,
            2627: PrimaryKeyViolation,
            8114: FailedTypeConversion,
            102: IncorrectSyntax,
            32: InvalidNumberParametersSupplied
        }.get(error_code, DefaultException)

    @classmethod
    def get_sql_server_error_code(cls, pyodbc_code: str, message: str) -> int:
        """
        Parses error message raised by PyODBC and return SQL Server Error Code

        Looks for the following pattern:
            (544) (SQLExecDirectW) -> 544

        Args:
            pyodbc_error_message (str): Error string raised by PyODBC

        Returns:
            (int) - SQL Server Error Code
        """

        if pyodbc_code == cls.pyodbc_error_code:
            return 32
        else:
            m = re.search(cls.sql_error_code_pattern, message)
            if m:
                return int(m.group('sql_server_error_code'))
            else:
                raise ValueError(f"Error raised is not from SQL Server: {message}")

    @classmethod
    def build_pyodbc_exception(cls, pyodbc_exception: Exception):
        pyodbc_code = cls.get_pyodbc_code(pyodbc_exception)
        error_message = cls.get_message(pyodbc_exception)
        error_code = cls.get_sql_server_error_code(pyodbc_code, error_message)
        exception = cls.get_exception(error_code)
        raise exception(error_message)
class IdentityInsertNull(PyODBCError):
    """
    Handle specific PyODBC error related to Null Value Inserted on Identity Column
    """

    def __init__(self, pyodbc_error_message):
        super().__init__(pyodbc_error_message)
        m = re.search(self.table_pattern, self.error_message)
        self.table_name = m.group('table_name')
        m = re.search(self.column_pattern, self.error_message)
        self.column_name = m.group('column_name')


class IdentityInsertSetToOff(PyODBCError):
    """
    Handle specific PyODBC error related to Identity Not Set to On/Off
    """

    def __init__(self, pyodbc_error_message):
        super().__init__(pyodbc_error_message)
        m = re.search(self.table_pattern, self.error_message)
        self.table_name = m.group('table_name')


class FailedTypeConversion(PyODBCError):
    """
    Handle specific PyODBC error related to data type conversion
    """

    def __init__(self, pyodbc_error_message):
        super().__init__(pyodbc_error_message)


class PrimaryKeyViolation(PyODBCError):
    """
    Handle specific PyODBC error related to Primary Key Violation
    """

    def __init__(self, pyodbc_error_message):
        super().__init__(pyodbc_error_message)


class IncorrectSyntax(PyODBCError):
    """
    Handle specific PyODBC error related to incorrect syntax in query
    """

    def __init__(self, pyodbc_error_message):
        super().__init__(pyodbc_error_message)


class DefaultException(PyODBCError):
    """
    Handle default PyODBC errors
    """

    def __init__(self, pyodbc_error_message):
        super().__init__(pyodbc_error_message)

    def __str__(self) -> str:
        return f"{self.sql_server_error_id} - {self.error_message}"


class InvalidNumberParametersSupplied(Exception):
    def __init__(self, error_message) -> None:
        self.message = error_message

    def __str__(self) -> str:
        return self.message

Tramontane

Changed m = re.match() into m = re.search() and it worked perfectly. Thanks. – Pietje

Answer 4

In pyodbc 3.0.7, it works fine to catch pyodbc.ProgrammingError (and presumably the other error types, although I haven't tried). The contents of the error are still sort of cryptic, though, so it may be hard to do finer-grained handling of errors.

Hughs

Answer 5

Worked on me:

try:
    """Code to catch error"""

except Exception as e:
    sqlstate = e.args[0]
    if "42000" in sqlstate:
        print("This is ProgrammingError")
    elif "23000" in sqlstate:
        print("This is IntegrityError.")

The number is represents the error from pyodbc.

Shrum

Answer 6

this will give you more clear and readable error message when connecting to mssql using myodbc:

try:
    cnxn = pyodbc.connect(...)
except pyodbc.Error as ex:
    sqlstate = ex.args[1]
    sqlstate = sqlstate.split(".")
    print(sqlstate[-3])

Countercharge

Answer 7

Be carreful it looks like an pyodbc exceptions but it may not be:

• In my case I read pyodbc.IntegrityError in "(pyodbc.IntegrityError) ('23000', '[23000]..."
• In the python console I can read this: "sqlalchemy.exc.IntegrityError: (pyodbc.IntegrityError) ('23000', '[23000]..." It is actually a sqlalchemy.exc.IntegrityError exception, which wraps an pyodbc.IntegrityError

Digged