Fix improper error handling of execute_query method when QueryItem is used: #82

Signed-off-by: Tsuyoshi Hombashi <[email protected]>

Author: thombashi

simplesqlite/core.py

@@ -335,18 +335,21 @@ def execute_query(
         import time
 
         self.check_connection()
-        if typepy.is_null_string(query):
+
+        queryStr = str(query)
+
+        if typepy.is_null_string(queryStr):
             return None
 
         if self.debug_query or self.global_debug_query:
-            logger.debug(query)
+            logger.debug(queryStr)
 
         exec_start_time = time.time()
 
         assert self.connection  # to avoid type check error
 
         try:
-            result = self.connection.execute(str(query))
+            result = self.connection.execute(queryStr)
         except (sqlite3.OperationalError, sqlite3.IntegrityError) as e:
             if caller is None:
                 caller = logging.getLogger().findCaller()
@@ -358,18 +361,18 @@ def execute_query(
                         "failed to execute query at {:s}({:d}) {:s}".format(
                             file_path, line_no, func_name
                         ),
-                        f"  - query: {MultiByteStrDecoder(query).unicode_str}",
+                        f"  - query: {MultiByteStrDecoder(queryStr).unicode_str}",
                         f"  - msg:   {e}",
                         f"  - db:    {self.database_path}",
                     ]
                 )
             )
 
         if self.__is_profile:
-            self.__dict_query_count[str(query)] += 1
+            self.__dict_query_count[queryStr] += 1
 
             elapse_time = time.time() - exec_start_time
-            self.__dict_query_totalexectime[str(query)] += elapse_time
+            self.__dict_query_totalexectime[queryStr] += elapse_time
 
         return result
 

test/test_simplesqlite.py

@@ -151,6 +151,18 @@ def test_smoke_view(self, con):
         result = con.select(select="*", table_name="view1")
         assert result is not None
 
+    def test_execute_query(self, con):
+        from simplesqlite.query import Select
+
+        result = con.execute_query(Select("*", TEST_TABLE_NAME))
+        assert result is not None
+
+    def test_exception_query(self, con):
+        from simplesqlite.query import Select
+
+        with pytest.raises(OperationalError):
+            con.execute_query(Select("*", "invalid_table"))
+
     @pytest.mark.parametrize(
         ["attr", "table_name", "expected"],
         [