sqlite3 module expose sqlite error code and name in exceptions

python · pablogsal · Aug 30, 2021 · Apr 13, 2017 · Apr 13, 2017 · May 8, 2019
commit 83ab267abe9985664a5989488bd93e09aeefc963
diff --git a/Doc/includes/sqlite3/complete_statement.py b/Doc/includes/sqlite3/complete_statement.py
@@ -24,7 +24,10 @@
             if buffer.lstrip().upper().startswith("SELECT"):
                 print(cur.fetchall())
         except sqlite3.Error as e:
-            print("An error occurred:", e.args[0])
+            msg = str(e))
+            error_code = e.sqlite_errorcode
+            error_name = e.sqlite_name
+            print(f"Error {error_name} [Errno {error_code}]: {msg}")
         buffer = ""
 
 con.close()
@@ -270,6 +270,26 @@ Module functions and constants
    disable the feature again.
 
 
+.. exception:: Error
+
+   Raised to signal an error from the underlying SQLite library.
+
+   .. attribute:: sqlite_errorcode
+
+      The numeric error code from the `SQLite API
+      <http://www.sqlite.org/c3ref/c_abort.html>`_.
+
+      .. versionadded:: 3.7
+
+   .. attribute:: sqlite_errorname
+
+      The symbolic name of the numeric error code
+      from the `SQLite API
+      <http://www.sqlite.org/c3ref/c_abort.html>`_.
+
+      .. versionadded:: 3.7
+
+
 .. _sqlite3-connection-objects:
 
 Connection Objects

@@ -83,6 +83,14 @@ def CheckNotSupportedError(self):
                                    sqlite.DatabaseError),
                         "NotSupportedError is not a subclass of DatabaseError")
 
+    def CheckErrorCodeOnException(self):
+        with self.assertRaises(sqlite.Error) as cm:
+           db = sqlite.connect('/no/such/file/exists')
+        e = cm.exception
+        self.assertEqual(e.sqlite_errorcode, sqlite.SQLITE_CANTOPEN)
+        self.assertEqual(e.sqlite_errorname, "SQLITE_CANTOPEN")
+        self.assertEqual(str(e), "unable to open database file")
+
 class ConnectionTests(unittest.TestCase):
 
     def setUp(self):

@@ -271,13 +271,70 @@ struct _IntConstantPair {
 
 typedef struct _IntConstantPair IntConstantPair;
 
+/* sqlite API error codes */
+static const IntConstantPair _error_codes[] = {
+  {"SQLITE_OK", SQLITE_OK},
+  {"SQLITE_ERROR", SQLITE_ERROR},
+  {"SQLITE_INTERNAL", SQLITE_INTERNAL},
+  {"SQLITE_PERM", SQLITE_PERM},
+  {"SQLITE_ABORT", SQLITE_ABORT},
+  {"SQLITE_BUSY", SQLITE_BUSY},
+  {"SQLITE_LOCKED", SQLITE_LOCKED},
+  {"SQLITE_NOMEM", SQLITE_NOMEM},
+  {"SQLITE_READONLY", SQLITE_READONLY},
+  {"SQLITE_INTERRUPT", SQLITE_INTERRUPT},
+  {"SQLITE_IOERR", SQLITE_IOERR},
+  {"SQLITE_CORRUPT", SQLITE_CORRUPT},
+  {"SQLITE_NOTFOUND", SQLITE_NOTFOUND},
+  {"SQLITE_FULL", SQLITE_FULL},
+  {"SQLITE_CANTOPEN", SQLITE_CANTOPEN},
+  {"SQLITE_PROTOCOL", SQLITE_PROTOCOL},
+  {"SQLITE_EMPTY", SQLITE_EMPTY},
+  {"SQLITE_SCHEMA", SQLITE_SCHEMA},
+  {"SQLITE_TOOBIG", SQLITE_TOOBIG},
+  {"SQLITE_CONSTRAINT", SQLITE_CONSTRAINT},
+  {"SQLITE_MISMATCH", SQLITE_MISMATCH},
+  {"SQLITE_MISUSE", SQLITE_MISUSE},
+#ifdef SQLITE_NOLFS
+  {"SQLITE_NOLFS", SQLITE_NOLFS},
+#endif
+#ifdef SQLITE_AUTH
+  {"SQLITE_AUTH", SQLITE_AUTH},
+#endif
+#ifdef SQLITE_FORMAT
+  {"SQLITE_FORMAT", SQLITE_FORMAT},
+#endif
+#ifdef SQLITE_RANGE
+  {"SQLITE_RANGE", SQLITE_RANGE},
+#endif
+#ifdef SQLITE_NOTADB
+  {"SQLITE_NOTADB", SQLITE_NOTADB},
+#endif
+  {"SQLITE_DONE", SQLITE_DONE},
+  {"SQLITE_ROW", SQLITE_ROW},
+  {(char*)NULL, 0},
+  {"SQLITE_UNKNOWN", -1}
+};
+
+const char *sqlite3ErrName(int rc) {
+    int i;
+    for (i = 0; _error_codes[i].constant_name != 0; i++) {
+        if (_error_codes[i].constant_value == rc)
+          return _error_codes[i].constant_name;
+    }
+    // No error code matched.
+    return _error_codes[i+1].constant_name;
+}
+
 static const IntConstantPair _int_constants[] = {
     {"PARSE_DECLTYPES", PARSE_DECLTYPES},
     {"PARSE_COLNAMES", PARSE_COLNAMES},
 
-    {"SQLITE_OK", SQLITE_OK},
+    /* enumerated return values for sqlite3_set_authorizer() callback */
     {"SQLITE_DENY", SQLITE_DENY},
     {"SQLITE_IGNORE", SQLITE_IGNORE},
+
+    /* enumerated values for sqlite3_set_authorizer() callback */
     {"SQLITE_CREATE_INDEX", SQLITE_CREATE_INDEX},
     {"SQLITE_CREATE_TABLE", SQLITE_CREATE_TABLE},
     {"SQLITE_CREATE_TEMP_INDEX", SQLITE_CREATE_TEMP_INDEX},
@@ -342,6 +399,29 @@ static struct PyModuleDef _sqlite3module = {
         NULL
 };
 
+
+static int add_to_dict(PyObject *dict, const char *key, int value)
+{
+    int sawerror;
+    PyObject *value_obj = PyLong_FromLong(value);
+    PyObject *name = PyUnicode_FromString(key);
+
+    if (!value_obj || !name) {
+        Py_XDECREF(name);
+        Py_XDECREF(value_obj);
+        return 1;
+    }
+
+    sawerror = PyDict_SetItem(dict, name, value_obj) < 0;
+
+    Py_DECREF(value_obj);
+    Py_DECREF(name);
+
+    if (sawerror)
+        return 1;
+    return 0;
+}
+
 PyMODINIT_FUNC PyInit__sqlite3(void)
 {
     PyObject *module, *dict;
@@ -445,12 +525,16 @@ PyMODINIT_FUNC PyInit__sqlite3(void)
 
     /* Set integer constants */
     for (i = 0; _int_constants[i].constant_name != NULL; i++) {
-        tmp_obj = PyLong_FromLong(_int_constants[i].constant_value);
-        if (!tmp_obj) {
+        if (add_to_dict(dict, _int_constants[i].constant_name,
+                        _int_constants[i].constant_value) != 0)
+            goto error;
+    }
+
+    /* Set error constants */
+    for (i = 0; _error_codes[i].constant_name != 0; i++) {
+        if (add_to_dict(dict, _error_codes[i].constant_name,
+                        _error_codes[i].constant_value) != 0)
             goto error;
-        }
-        PyDict_SetItemString(dict, _int_constants[i].constant_name, tmp_obj);
-        Py_DECREF(tmp_obj);
     }
 
     if (!(tmp_obj = PyUnicode_FromString(PYSQLITE_VERSION))) {

@@ -48,6 +48,8 @@ extern PyObject* _pysqlite_converters;
 extern int _pysqlite_enable_callback_tracebacks;
 extern int pysqlite_BaseTypeAdapted;
 
+extern const char *sqlite3ErrName(int rc);
+
 #define PARSE_DECLTYPES 1
 #define PARSE_COLNAMES 2
 #endif
@@ -47,20 +47,21 @@ int pysqlite_step(sqlite3_stmt* statement, pysqlite_Connection* connection)
  */
 int _pysqlite_seterror(sqlite3* db, sqlite3_stmt* st)
 {
+    PyObject *exc_class;
     int errorcode = sqlite3_errcode(db);
 
     switch (errorcode)
     {
         case SQLITE_OK:
             PyErr_Clear();
-            break;
+            return errorcode;
         case SQLITE_INTERNAL:
         case SQLITE_NOTFOUND:
-            PyErr_SetString(pysqlite_InternalError, sqlite3_errmsg(db));
+            exc_class = pysqlite_InternalError;
             break;
         case SQLITE_NOMEM:
             (void)PyErr_NoMemory();
-            break;
+            return errorcode;
         case SQLITE_ERROR:
         case SQLITE_PERM:
         case SQLITE_ABORT:
@@ -74,26 +75,71 @@ int _pysqlite_seterror(sqlite3* db, sqlite3_stmt* st)
         case SQLITE_PROTOCOL:
         case SQLITE_EMPTY:
         case SQLITE_SCHEMA:
-            PyErr_SetString(pysqlite_OperationalError, sqlite3_errmsg(db));
+            exc_class = pysqlite_OperationalError;
             break;
         case SQLITE_CORRUPT:
-            PyErr_SetString(pysqlite_DatabaseError, sqlite3_errmsg(db));
+            exc_class = pysqlite_DatabaseError;
             break;
         case SQLITE_TOOBIG:
-            PyErr_SetString(pysqlite_DataError, sqlite3_errmsg(db));
+            exc_class = pysqlite_DataError;
             break;
         case SQLITE_CONSTRAINT:
         case SQLITE_MISMATCH:
-            PyErr_SetString(pysqlite_IntegrityError, sqlite3_errmsg(db));
+            exc_class = pysqlite_IntegrityError;
             break;
         case SQLITE_MISUSE:
-            PyErr_SetString(pysqlite_ProgrammingError, sqlite3_errmsg(db));
+            exc_class = pysqlite_ProgrammingError;
             break;
         default:
-            PyErr_SetString(pysqlite_DatabaseError, sqlite3_errmsg(db));
+            exc_class = pysqlite_DatabaseError;
             break;
     }
 
+    /* Create and set the exception. */
+    {
+        const char *error_msg;
+        const char *error_name;
+        PyObject *exc = NULL;
+        PyObject *args = NULL;
+        PyObject *py_code = NULL;
+        PyObject *py_name = NULL;
+
+        error_name = sqlite3ErrName(errorcode);
+
+        error_msg = sqlite3_errmsg(db);
+
+        args = Py_BuildValue("(s)", error_msg);
+        if (!args)
+            goto error;
+
+        exc = PyObject_Call(exc_class, args, NULL);
+        if (!exc)
+            goto error;
+
+        py_code = Py_BuildValue("i", errorcode);
+        if (!py_code)
+            goto error;
+
+        if (PyObject_SetAttrString(exc, "sqlite_errorcode", py_code) < 0)
+            goto error;
+
+        py_name = Py_BuildValue("s", error_name);
+        if (!py_name)
+            goto error;
+
+        if (PyObject_SetAttrString(exc, "sqlite_errorname", py_name) < 0)
+            goto error;
+
+        PyErr_SetObject((PyObject *) Py_TYPE(exc), exc);
+
+    error:
+        Py_XDECREF(py_code);
+        Py_XDECREF(py_name);
+        Py_XDECREF(args);
+        Py_XDECREF(exc);
+    }
+
+
     return errorcode;
 }