Merge pull request #12 from AnswerDotAI/remove-operational-error

Remove OperationError

apswutils/db.py

@@ -1,13 +1,13 @@
 # This file is from sqlite-utils and copyright and license is the same as that project
 __all__ = ['Database', 'Queryable', 'Table', 'View']
 
-from .utils import chunks, hash_record, OperationalError, suggest_column_types, types_for_column_types, column_affinity, find_spatialite
+from .utils import chunks, hash_record, suggest_column_types, types_for_column_types, column_affinity, find_spatialite
 from collections import namedtuple
 from collections.abc import Mapping
 from typing import cast, Any, Callable, Dict, Generator, Iterable, Union, Optional, List, Tuple, Iterator
 from functools import cache
 import contextlib, datetime, decimal, inspect, itertools, json, os, pathlib, re, secrets, textwrap, binascii, uuid, logging
-import apsw.ext, apsw.bestpractice
+import apsw, apsw.ext, apsw.bestpractice
 
 logger = logging.getLogger('apsw')
 logger.setLevel(logging.ERROR)
@@ -684,7 +684,7 @@ def cached_counts(self, tables: Optional[Iterable[str]] = None) -> Dict[str, int
             sql += " where [table] in ({})".format(", ".join("?" for table in tables))
         try:
             return {r[0]: r[1] for r in self.execute(sql, tables).fetchall()}
-        except OperationalError:
+        except apsw.Error:
             return {}
 
     def reset_counts(self):
@@ -2118,7 +2118,7 @@ def create_index(
             try:
                 self.db.execute(sql)
                 break
-            except OperationalError as e:
+            except apsw.SQLError as e:
                 # find_unique_name=True - try again if 'index ... already exists'
                 arg = e.args[0]
                 if (
@@ -2769,7 +2769,7 @@ def update(
 
             for row in cursor:
                 self.result.append(dict(zip(columns, row)))
-        except OperationalError as e:
+        except apsw.SQLError as e:
             if alter and (" column" in e.args[0]):
                 # Attempt to add any missing columns, then try again
                 self.add_missing_columns([updates])
@@ -2934,7 +2934,7 @@ def insert_chunk(
                 except apsw.ExecutionCompleteError: continue
                 for row in cursor:
                     records.append(dict(zip(columns, row)))
-            except OperationalError as e:
+            except apsw.SQLError as e:
                 if alter and (" column" in e.args[0]):
                     # Attempt to add any missing columns, then try again
                     self.add_missing_columns(chunk)

apswutils/utils.py

@@ -13,10 +13,6 @@
 from typing import Dict, cast, BinaryIO, Iterable, Optional, Tuple, Type
 import apsw
 
-# TODO: Replace use of OperationalError with more explicit apsw errors
-#       In order to keep this PR minimal, we use OperationalError  as a shim for apsw.Error
-OperationalError = apsw.Error
-
 SPATIALITE_PATHS = (
     "/usr/lib/x86_64-linux-gnu/mod_spatialite.so",
     "/usr/lib/aarch64-linux-gnu/mod_spatialite.so",

tests/test_create.py

@@ -4,7 +4,6 @@
     DescIndex,
     AlterError,
     NoObviousTable,
-    OperationalError,
     ForeignKey,
     Table,
     View,
@@ -717,7 +716,7 @@ def test_columns_not_in_first_record_should_not_cause_batch_to_be_too_large(fres
         fresh_db["too_many_columns"].insert_all(
             records, alter=True, batch_size=batch_size
         )
-    except OperationalError:
+    except apsw.SQLError:
         raise
 
 
@@ -817,7 +816,7 @@ def test_create_index_find_unique_name(fresh_db):
     table.insert({"id": 1})
     table.create_index(["id"])
     # Without find_unique_name should error
-    with pytest.raises(OperationalError, match="index idx_t_id already exists"):
+    with pytest.raises(apsw.SQLError, match="index idx_t_id already exists"):
         table.create_index(["id"])
     # With find_unique_name=True it should work
     table.create_index(["id"], find_unique_name=True)

tests/test_create_view.py

@@ -1,5 +1,5 @@
 import pytest
-from apswutils.utils import OperationalError
+import apsw
 
 
 def test_create_view(fresh_db):
@@ -10,7 +10,7 @@ def test_create_view(fresh_db):
 
 def test_create_view_error(fresh_db):
     fresh_db.create_view("bar", "select 1 + 1")
-    with pytest.raises(OperationalError):
+    with pytest.raises(apsw.SQLError):
         fresh_db.create_view("bar", "select 1 + 2")
 
 

tests/test_transform.py

@@ -1,5 +1,4 @@
 from apswutils.db import ForeignKey
-from apswutils.utils import OperationalError
 import pytest
 import apsw
 from collections.abc import Mapping