Squash abstract database into simply base database

Author: Sergey Vasilyev

data_diff/abcs/database_types.py

@@ -268,76 +268,6 @@ def to_comparable(self, value: str, coltype: ColType) -> str:
         """Ensure that the expression is comparable in ``IS DISTINCT FROM``."""
 
 
-from typing import TypeVar, Generic
-
-T_Dialect = TypeVar("T_Dialect", bound=AbstractDialect)
-
-
-class AbstractDatabase(Generic[T_Dialect]):
-    @property
-    @abstractmethod
-    def dialect(self) -> T_Dialect:
-        "The dialect of the database. Used internally by Database, and also available publicly."
-
-    @classmethod
-    @abstractmethod
-    def load_mixins(cls, *abstract_mixins) -> type:
-        "Extend the dialect with a list of mixins that implement the given abstract mixins."
-
-    @property
-    @abstractmethod
-    def CONNECT_URI_HELP(self) -> str:
-        "Example URI to show the user in help and error messages"
-
-    @property
-    @abstractmethod
-    def CONNECT_URI_PARAMS(self) -> List[str]:
-        "List of parameters given in the path of the URI"
-
-    @abstractmethod
-    def _query(self, sql_code: str) -> list:
-        "Send query to database and return result"
-
-    @abstractmethod
-    def query_table_schema(self, path: DbPath) -> Dict[str, tuple]:
-        """Query the table for its schema for table in 'path', and return {column: tuple}
-        where the tuple is (table_name, col_name, type_repr, datetime_precision?, numeric_precision?, numeric_scale?)
-
-        Note: This method exists instead of select_table_schema(), just because not all databases support
-              accessing the schema using a SQL query.
-        """
-
-    @abstractmethod
-    def select_table_unique_columns(self, path: DbPath) -> str:
-        "Provide SQL for selecting the names of unique columns in the table"
-
-    @abstractmethod
-    def query_table_unique_columns(self, path: DbPath) -> List[str]:
-        """Query the table for its unique columns for table in 'path', and return {column}"""
-
-    @abstractmethod
-    def _process_table_schema(
-        self, path: DbPath, raw_schema: Dict[str, tuple], filter_columns: Sequence[str], where: str = None
-    ):
-        """Process the result of query_table_schema().
-
-        Done in a separate step, to minimize the amount of processed columns.
-        Needed because processing each column may:
-        * throw errors and warnings
-        * query the database to sample values
-
-        """
-
-    @abstractmethod
-    def close(self):
-        "Close connection(s) to the database instance. Querying will stop functioning."
-
-    @property
-    @abstractmethod
-    def is_autocommit(self) -> bool:
-        "Return whether the database autocommits changes. When false, COMMIT statements are skipped."
-
-
 class AbstractTable(ABC):
     @abstractmethod
     def select(self, *exprs, distinct=False, **named_exprs) -> "AbstractTable":

data_diff/databases/base.py

@@ -1,3 +1,4 @@
+import abc
 import functools
 from dataclasses import field
 from datetime import datetime
@@ -31,7 +32,6 @@
     Root, TableAlias, TableOp, TablePath, \
     TimeTravel, TruncateTable, UnaryOp, WhenThen, _ResolveColumn
 from data_diff.abcs.database_types import (
-    AbstractDatabase,
     Array,
     Struct,
     AbstractDialect,
@@ -92,7 +92,7 @@ class Compiler(AbstractCompiler):
     # Database is needed to normalize tables. Dialect is needed for recursive compilations.
     # In theory, it is many-to-many relations: e.g. a generic ODBC driver with multiple dialects.
     # In practice, we currently bind the dialects to the specific database classes.
-    database: AbstractDatabase
+    database: "Database"
 
     in_select: bool = False  # Compilation runtime flag
     in_join: bool = False  # Compilation runtime flag
@@ -789,7 +789,7 @@ def __getitem__(self, i):
         return self.rows[i]
 
 
-class Database(AbstractDatabase[T]):
+class Database(abc.ABC):
     """Base abstract class for databases.
 
     Used for providing connection code and implementation specific SQL utilities.
@@ -898,6 +898,12 @@ def select_table_schema(self, path: DbPath) -> str:
         )
 
     def query_table_schema(self, path: DbPath) -> Dict[str, tuple]:
+        """Query the table for its schema for table in 'path', and return {column: tuple}
+        where the tuple is (table_name, col_name, type_repr, datetime_precision?, numeric_precision?, numeric_scale?)
+
+        Note: This method exists instead of select_table_schema(), just because not all databases support
+              accessing the schema using a SQL query.
+        """
         rows = self.query(self.select_table_schema(path), list)
         if not rows:
             raise RuntimeError(f"{self.name}: Table '{'.'.join(path)}' does not exist, or has no columns")
@@ -907,6 +913,7 @@ def query_table_schema(self, path: DbPath) -> Dict[str, tuple]:
         return d
 
     def select_table_unique_columns(self, path: DbPath) -> str:
+        "Provide SQL for selecting the names of unique columns in the table"
         schema, name = self._normalize_table_path(path)
 
         return (
@@ -916,6 +923,7 @@ def select_table_unique_columns(self, path: DbPath) -> str:
         )
 
     def query_table_unique_columns(self, path: DbPath) -> List[str]:
+        """Query the table for its unique columns for table in 'path', and return {column}"""
         if not self.SUPPORTS_UNIQUE_CONSTAINT:
             raise NotImplementedError("This database doesn't support 'unique' constraints")
         res = self.query(self.select_table_unique_columns(path), List[str])
@@ -924,6 +932,14 @@ def query_table_unique_columns(self, path: DbPath) -> List[str]:
     def _process_table_schema(
         self, path: DbPath, raw_schema: Dict[str, tuple], filter_columns: Sequence[str] = None, where: str = None
     ):
+        """Process the result of query_table_schema().
+
+        Done in a separate step, to minimize the amount of processed columns.
+        Needed because processing each column may:
+        * throw errors and warnings
+        * query the database to sample values
+
+        """
         if filter_columns is None:
             filtered_schema = raw_schema
         else:
@@ -1017,12 +1033,37 @@ def _query_conn(self, conn, sql_code: Union[str, ThreadLocalInterpreter]) -> Que
         return apply_query(callback, sql_code)
 
     def close(self):
+        "Close connection(s) to the database instance. Querying will stop functioning."
         self.is_closed = True
         return super().close()
 
     def list_tables(self, tables_like, schema=None):
         return self.query(self.dialect.list_tables(schema or self.default_schema, tables_like))
 
+    @property
+    @abstractmethod
+    def dialect(self) -> BaseDialect:
+        "The dialect of the database. Used internally by Database, and also available publicly."
+
+    @property
+    @abstractmethod
+    def CONNECT_URI_HELP(self) -> str:
+        "Example URI to show the user in help and error messages"
+
+    @property
+    @abstractmethod
+    def CONNECT_URI_PARAMS(self) -> List[str]:
+        "List of parameters given in the path of the URI"
+
+    @abstractmethod
+    def _query(self, sql_code: str) -> list:
+        "Send query to database and return result"
+
+    @property
+    @abstractmethod
+    def is_autocommit(self) -> bool:
+        "Return whether the database autocommits changes. When false, COMMIT statements are skipped."
+
 
 class ThreadedDatabase(Database):
     """Access the database through singleton threads.

data_diff/schema.py

@@ -1,14 +1,15 @@
 import logging
 
+from data_diff import Database
 from data_diff.utils import CaseAwareMapping, CaseInsensitiveDict, CaseSensitiveDict
-from data_diff.abcs.database_types import AbstractDatabase, DbPath
+from data_diff.abcs.database_types import DbPath
 
 logger = logging.getLogger("schema")
 
 Schema = CaseAwareMapping
 
 
-def create_schema(db: AbstractDatabase, table_path: DbPath, schema: dict, case_sensitive: bool) -> CaseAwareMapping:
+def create_schema(db: Database, table_path: DbPath, schema: dict, case_sensitive: bool) -> CaseAwareMapping:
     logger.debug(f"[{db.name}] Schema = {schema}")
 
     if case_sensitive: