Merge pull request #6731 from psa/sqlite-fine-detail

Add SQLite detailed error codes to error messages

Author: wxtim

changes.d/6730.feat.md

@@ -0,0 +1 @@
+Add SQLite detailed error codes to error logs

cylc/flow/rundb.py

@@ -505,22 +505,46 @@ def execute_queued_items(self):
         # something went wrong
         # (includes DB file not found, transaction processing issue, db locked)
         except sqlite3.Error as e:
+            # Detailed error codes are only available for python >= 3.11
+            if hasattr(e, "sqlite_errorcode"):
+                error_code = str(e.sqlite_errorcode)
+            else:
+                error_code = "Not available"
+
+            if hasattr(e, "sqlite_errorname"):
+                error_name = e.sqlite_errorname
+            else:
+                error_name = "Not available"
+
             if not self.is_public:
                 # incase this isn't a filesystem issue, log the statements
                 # which make up the transaction to assist debug
                 LOG.error(
-                    'An error occurred when writing to the database,'
+                    'An error occurred when writing to the database %(file)s,'
                     ' this is probably a filesystem issue.'
-                    f' The attempted transaction was:\n{pformat(sql_queue)}'
+                    ' The error was: %(error)s'
+                    ' SQLite error code: %(error_code)s'
+                    ' SQLite error name: %(error_name)s'
+                    ' The attempted transaction was:\n %(transaction)s' % {
+                        "file": self.db_file_name,
+                        "error": str(e),
+                        "error_code": error_code,
+                        "error_name": error_name,
+                        "transaction": pformat(sql_queue)
+                    }
                 )
                 raise
             self.n_tries += 1
             LOG.warning(
                 "%(file)s: write attempt (%(attempt)d)"
-                " did not complete: %(error)s\n" % {
+                " did not complete: %(error)s\n"
+                " SQLite error code: %(error_code)s\n"
+                " SQLite error name: %(error_name)s" % {
                     "file": self.db_file_name,
                     "attempt": self.n_tries,
-                    "error": str(e)
+                    "error": str(e),
+                    "error_code": error_code,
+                    "error_name": error_name
                 }
             )
             if self.conn is not None:
@@ -565,15 +589,15 @@ def _execute_stmt(self, stmt, stmt_args_list):
         try:
             self.connect()
             self.conn.executemany(stmt, stmt_args_list)
-        except sqlite3.Error:
+        except sqlite3.Error as e:
             if not self.is_public:
                 raise
             if cylc.flow.flags.verbosity > 1:
                 traceback.print_exc()
             err_log = (
                 "cannot execute database statement:\n"
-                "file=%(file)s:\nstmt=%(stmt)s"
-            ) % {"file": self.db_file_name, "stmt": stmt}
+                "file=%(file)s:\nstmt=%(stmt)s\nerror=%(error)s"
+            ) % {"file": self.db_file_name, "stmt": stmt, "error": str(e)}
             for i, stmt_args in enumerate(stmt_args_list):
                 err_log += ("\nstmt_args[%(i)d]=%(stmt_args)s" % {
                     "i": i, "stmt_args": stmt_args})

tests/unit/test_rundb.py

@@ -25,6 +25,7 @@
 )
 import unittest
 from unittest import mock
+import sys
 
 import pytest
 
@@ -132,6 +133,13 @@ def test_operational_error(tmp_path, caplog):
     assert 'DELETE FROM task_jobs' in message
     assert 'INSERT OR REPLACE INTO task_jobs' in message
     assert 'UPDATE task_jobs' in message
+    assert 'The error was: no such table: task_jobs' in message
+    if sys.version_info.major == 3 and sys.version_info.minor >= 11:
+        assert 'SQLite error code: 1' in message
+        assert 'SQLite error name: SQLITE_ERROR' in message
+    else:
+        assert 'SQLite error code: Not available' in message
+        assert 'SQLite error name: Not available' in message
 
 
 def test_table_creation(tmp_path: Path):