Merge pull request datafold#329 from datafold/dec2_cleanup

Dec2 cleanup

Author: erezsh

data_diff/__init__.py

@@ -9,6 +9,7 @@
 from .table_segment import TableSegment
 from .utils import eval_name_template
 
+
 def connect_to_table(
     db_info: Union[str, dict],
     table_name: Union[DbPath, str],

data_diff/__main__.py

@@ -81,8 +81,8 @@ def __init__(self, **kwargs):
     def write_usage(self, prog: str, args: str = "", prefix: Optional[str] = None) -> None:
         self.write(f"data-diff v{__version__} - efficiently diff rows across database tables.\n\n")
         self.write("Usage:\n")
-        self.write(f"  * In-db diff:    {prog} <database1> <table1> <table2> [OPTIONS]\n")
-        self.write(f"  * Cross-db diff: {prog} <database1> <table1> <database2> <table2> [OPTIONS]\n")
+        self.write(f"  * In-db diff:    {prog} <database_a> <table_a> <table_b> [OPTIONS]\n")
+        self.write(f"  * Cross-db diff: {prog} <database_a> <table_a> <database_b> <table_b> [OPTIONS]\n")
         self.write(f"  * Using config:  {prog} --conf PATH [--run NAME] [OPTIONS]\n")
 
 

data_diff/databases/_connect.py

@@ -1,6 +1,7 @@
-from data_diff.sqeleton.databases import Connect, Database
 import logging
 
+from data_diff.sqeleton.databases import Connect
+
 from .postgresql import PostgreSQL
 from .mysql import MySQL
 from .oracle import Oracle

data_diff/diff_tables.py

@@ -7,7 +7,7 @@
 from enum import Enum
 from contextlib import contextmanager
 from operator import methodcaller
-from typing import Dict, Iterable, Tuple, Iterator, Optional
+from typing import Dict, Tuple, Iterator, Optional
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from runtype import dataclass

data_diff/info_tree.py

@@ -1,5 +1,4 @@
-from typing import Tuple, List, Dict
-from itertools import chain
+from typing import List, Dict
 
 from runtype import dataclass
 

data_diff/sqeleton/abcs/database_types.py

@@ -162,23 +162,19 @@ def ROUNDS_ON_PREC_LOSS(self) -> bool:
     @abstractmethod
     def quote(self, s: str):
         "Quote SQL name"
-        ...
 
     @abstractmethod
     def concat(self, items: List[str]) -> str:
         "Provide SQL for concatenating a bunch of columns into a string"
-        ...
 
     @abstractmethod
     def is_distinct_from(self, a: str, b: str) -> str:
         "Provide SQL for a comparison where NULL = NULL is true"
-        ...
 
     @abstractmethod
     def to_string(self, s: str) -> str:
         # TODO rewrite using cast_to(x, str)
         "Provide SQL for casting a column to string"
-        ...
 
     @abstractmethod
     def random(self) -> str:
@@ -191,17 +187,14 @@ def current_timestamp(self) -> str:
     @abstractmethod
     def offset_limit(self, offset: Optional[int] = None, limit: Optional[int] = None):
         "Provide SQL fragment for limit and offset inside a select"
-        ...
 
     @abstractmethod
     def explain_as_text(self, query: str) -> str:
         "Provide SQL for explaining a query, returned as table(varchar)"
-        ...
 
     @abstractmethod
     def timestamp_value(self, t: datetime) -> str:
         "Provide SQL for the given timestamp value"
-        ...
 
     @abstractmethod
     def set_timezone_to_utc(self) -> str:
@@ -239,7 +232,6 @@ def CONNECT_URI_PARAMS(self) -> List[str]:
     @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]:
@@ -249,17 +241,14 @@ def query_table_schema(self, path: DbPath) -> Dict[str, tuple]:
         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(
@@ -277,21 +266,10 @@ def _process_table_schema(
     @abstractmethod
     def parse_table_name(self, name: str) -> DbPath:
         "Parse the given table name into a DbPath"
-        ...
 
     @abstractmethod
     def close(self):
         "Close connection(s) to the database instance. Querying will stop functioning."
-        ...
-
-    @property
-    @abstractmethod
-    def is_closed(self) -> bool:
-        "Return whether or not the connection has been closed"
-
-    @abstractmethod
-    def _normalize_table_path(self, path: DbPath) -> DbPath:
-        ...
 
     @property
     @abstractmethod

data_diff/sqeleton/abcs/mixins.py

@@ -14,7 +14,6 @@ def normalize_timestamp(self, value: str, coltype: TemporalType) -> str:
 
         Precision of dates should be rounded up/down according to coltype.rounds
         """
-        ...
 
     @abstractmethod
     def normalize_number(self, value: str, coltype: FractionalType) -> str:
@@ -35,17 +34,8 @@ def normalize_number(self, value: str, coltype: FractionalType) -> str:
         it's the same as ``numeric_scale``, and for floats, who use binary precision,
         it can be calculated as ``log10(2**numeric_precision)``.
         """
-        ...
 
-    @abstractmethod
-    def normalize_uuid(self, value: str, coltype: ColType_UUID) -> str:
-        """Creates an SQL expression, that converts 'value' to a normalized uuid.
-
-        i.e. just makes sure there is no trailing whitespace.
-        """
-        ...
-
-    def normalize_boolean(self, value: str, coltype: Boolean) -> str:
+    def normalize_boolean(self, value: str, _coltype: Boolean) -> str:
         """Creates an SQL expression, that converts 'value' to either '0' or '1'."""
         return self.to_string(value)
 
@@ -88,7 +78,6 @@ class AbstractMixin_MD5(ABC):
     @abstractmethod
     def md5_as_int(self, s: str) -> str:
         "Provide SQL for computing md5 and returning an int"
-        ...
 
 
 class AbstractMixin_Schema(ABC):

data_diff/sqeleton/databases/base.py

@@ -320,12 +320,12 @@ def enable_interactive(self):
 
     def select_table_schema(self, path: DbPath) -> str:
         """Provide SQL for selecting the table schema as (name, type, date_prec, num_prec)"""
-        schema, table = self._normalize_table_path(path)
+        schema, name = self._normalize_table_path(path)
 
         return (
             "SELECT column_name, data_type, datetime_precision, numeric_precision, numeric_scale "
             "FROM information_schema.columns "
-            f"WHERE table_name = '{table}' AND table_schema = '{schema}'"
+            f"WHERE table_name = '{name}' AND table_schema = '{schema}'"
         )
 
     def query_table_schema(self, path: DbPath) -> Dict[str, tuple]:
@@ -338,12 +338,12 @@ def query_table_schema(self, path: DbPath) -> Dict[str, tuple]:
         return d
 
     def select_table_unique_columns(self, path: DbPath) -> str:
-        schema, table = self._normalize_table_path(path)
+        schema, name = self._normalize_table_path(path)
 
         return (
             "SELECT column_name "
             "FROM information_schema.key_column_usage "
-            f"WHERE table_name = '{table}' AND table_schema = '{schema}'"
+            f"WHERE table_name = '{name}' AND table_schema = '{schema}'"
         )
 
     def query_table_unique_columns(self, path: DbPath) -> List[str]:
@@ -431,7 +431,7 @@ def _query_cursor(self, c, sql_code: str):
             c.execute(sql_code)
             if sql_code.lower().startswith(("select", "explain", "show")):
                 return c.fetchall()
-        except Exception as e:
+        except Exception as _e:
             # logger.exception(e)
             # logger.error(f'Caused by SQL: {sql_code}')
             raise
@@ -478,7 +478,6 @@ def _query_in_worker(self, sql_code: Union[str, ThreadLocalInterpreter]):
     @abstractmethod
     def create_connection(self):
         "Return a connection instance, that supports the .cursor() method."
-        ...
 
     def close(self):
         super().close()

data_diff/sqeleton/databases/bigquery.py

@@ -49,7 +49,7 @@ def normalize_timestamp(self, value: str, coltype: TemporalType) -> str:
     def normalize_number(self, value: str, coltype: FractionalType) -> str:
         return f"format('%.{coltype.precision}f', {value})"
 
-    def normalize_boolean(self, value: str, coltype: Boolean) -> str:
+    def normalize_boolean(self, value: str, _coltype: Boolean) -> str:
         return self.to_string(f"cast({value} as int)")
 
 
@@ -145,11 +145,12 @@ def close(self):
         self._client.close()
 
     def select_table_schema(self, path: DbPath) -> str:
-        schema, table = self._normalize_table_path(path)
+        schema, name = self._normalize_table_path(path)
 
         return (
-            f"SELECT column_name, data_type, 6 as datetime_precision, 38 as numeric_precision, 9 as numeric_scale FROM {schema}.INFORMATION_SCHEMA.COLUMNS "
-            f"WHERE table_name = '{table}' AND table_schema = '{schema}'"
+            "SELECT column_name, data_type, 6 as datetime_precision, 38 as numeric_precision, 9 as numeric_scale "
+            f"FROM {schema}.INFORMATION_SCHEMA.COLUMNS "
+            f"WHERE table_name = '{name}' AND table_schema = '{schema}'"
         )
 
     def query_table_unique_columns(self, path: DbPath) -> List[str]:

data_diff/sqeleton/databases/clickhouse.py

@@ -8,7 +8,6 @@
     ThreadedDatabase,
     import_helper,
     ConnectError,
-    DbTime,
 )
 from ..abcs.database_types import (
     ColType,
@@ -23,7 +22,9 @@
 )
 from ..abcs.mixins import AbstractMixin_MD5, AbstractMixin_NormalizeValue
 
-DEFAULT_DATABASE = "default"    # https://clickhouse.com/docs/en/operations/server-configuration-parameters/settings/#default-database
+# https://clickhouse.com/docs/en/operations/server-configuration-parameters/settings/#default-database
+DEFAULT_DATABASE = "default"
+
 
 @import_helper("clickhouse")
 def import_clickhouse():

data_diff/sqeleton/databases/connect.py

@@ -1,7 +1,7 @@
-from typing import Type, List, Optional, Union, Dict
+from typing import Type, Optional, Union, Dict
 from itertools import zip_longest
-import dsnparse
 from contextlib import suppress
+import dsnparse
 
 from runtype import dataclass
 
@@ -24,24 +24,17 @@
 @dataclass
 class MatchUriPath:
     database_cls: Type[Database]
-    params: List[str]
-    kwparams: List[str] = []
-    help_str: str = "<unspecified>"
-
-    def __post_init__(self):
-        assert self.params == self.database_cls.CONNECT_URI_PARAMS, self.params
-        assert self.help_str == self.database_cls.CONNECT_URI_HELP, "\n%s\n%s" % (
-            self.help_str,
-            self.database_cls.CONNECT_URI_HELP,
-        )
-        assert self.kwparams == self.database_cls.CONNECT_URI_KWPARAMS
 
     def match_path(self, dsn):
+        help_str = self.database_cls.CONNECT_URI_HELP
+        params = self.database_cls.CONNECT_URI_PARAMS
+        kwparams = self.database_cls.CONNECT_URI_KWPARAMS
+
         dsn_dict = dict(dsn.query)
         matches = {}
-        for param, arg in zip_longest(self.params, dsn.paths):
+        for param, arg in zip_longest(params, dsn.paths):
             if param is None:
-                raise ValueError(f"Too many parts to path. Expected format: {self.help_str}")
+                raise ValueError(f"Too many parts to path. Expected format: {help_str}")
 
             optional = param.endswith("?")
             param = param.rstrip("?")
@@ -51,26 +44,26 @@ def match_path(self, dsn):
                     arg = dsn_dict.pop(param)
                 except KeyError:
                     if not optional:
-                        raise ValueError(f"URI must specify '{param}'. Expected format: {self.help_str}")
+                        raise ValueError(f"URI must specify '{param}'. Expected format: {help_str}")
 
                     arg = None
 
             assert param and param not in matches
             matches[param] = arg
 
-        for param in self.kwparams:
+        for param in kwparams:
             try:
                 arg = dsn_dict.pop(param)
             except KeyError:
-                raise ValueError(f"URI must specify '{param}'. Expected format: {self.help_str}")
+                raise ValueError(f"URI must specify '{param}'. Expected format: {help_str}")
 
             assert param and arg and param not in matches, (param, arg, matches.keys())
             matches[param] = arg
 
         for param, value in dsn_dict.items():
             if param in matches:
                 raise ValueError(
-                    f"Parameter '{param}' already provided as positional argument. Expected format: {self.help_str}"
+                    f"Parameter '{param}' already provided as positional argument. Expected format: {help_str}"
                 )
 
             matches[param] = value
@@ -99,10 +92,7 @@ class Connect:
 
     def __init__(self, database_by_scheme: Dict[str, Database]):
         self.database_by_scheme = database_by_scheme
-        self.match_uri_path = {
-            name: MatchUriPath(cls, cls.CONNECT_URI_PARAMS, cls.CONNECT_URI_KWPARAMS, help_str=cls.CONNECT_URI_HELP)
-            for name, cls in database_by_scheme.items()
-        }
+        self.match_uri_path = {name: MatchUriPath(cls) for name, cls in database_by_scheme.items()}
         self.conn_cache = WeakCache()
 
     def connect_to_uri(self, db_uri: str, thread_count: Optional[int] = 1) -> Database:

data_diff/sqeleton/databases/databricks.py

@@ -48,7 +48,7 @@ def normalize_number(self, value: str, coltype: NumericType) -> str:
             value = f"format_number({value}, {coltype.precision})"
         return f"replace({self.to_string(value)}, ',', '')"
 
-    def normalize_boolean(self, value: str, coltype: Boolean) -> str:
+    def normalize_boolean(self, value: str, _coltype: Boolean) -> str:
         return self.to_string(f"cast ({value} as int)")
 
 

data_diff/sqeleton/databases/duckdb.py

@@ -50,7 +50,7 @@ def normalize_timestamp(self, value: str, coltype: TemporalType) -> str:
     def normalize_number(self, value: str, coltype: FractionalType) -> str:
         return self.to_string(f"{value}::DECIMAL(38, {coltype.precision})")
 
-    def normalize_boolean(self, value: str, coltype: Boolean) -> str:
+    def normalize_boolean(self, value: str, _coltype: Boolean) -> str:
         return self.to_string(f"{value}::INTEGER")
 
 
@@ -114,7 +114,7 @@ def set_timezone_to_utc(self) -> str:
 
 class DuckDB(Database):
     dialect = Dialect()
-    SUPPORTS_UNIQUE_CONSTAINT = False  # XXX Temporary, until implemented
+    SUPPORTS_UNIQUE_CONSTAINT = False  # Temporary, until we implement it
     default_schema = "main"
     CONNECT_URI_HELP = "duckdb://<database>@<dbpath>"
     CONNECT_URI_PARAMS = ["database", "dbpath"]

data_diff/sqeleton/databases/oracle.py

@@ -17,7 +17,7 @@
 from ..abcs.mixins import AbstractMixin_MD5, AbstractMixin_NormalizeValue, AbstractMixin_Schema
 from ..abcs import Compilable
 from ..queries import this, table, SKIP
-from .base import BaseDialect, ThreadedDatabase, import_helper, ConnectError, QueryError, Mixin_Schema
+from .base import BaseDialect, ThreadedDatabase, import_helper, ConnectError, QueryError
 from .base import TIMESTAMP_PRECISION_POS
 
 SESSION_TIME_ZONE = None  # Changed by the tests
@@ -182,9 +182,9 @@ def _query_cursor(self, c, sql_code: str):
             raise QueryError(e)
 
     def select_table_schema(self, path: DbPath) -> str:
-        schema, table = self._normalize_table_path(path)
+        schema, name = self._normalize_table_path(path)
 
         return (
             f"SELECT column_name, data_type, 6 as datetime_precision, data_precision as numeric_precision, data_scale as numeric_scale"
-            f" FROM ALL_TAB_COLUMNS WHERE table_name = '{table}' AND owner = '{schema}'"
+            f" FROM ALL_TAB_COLUMNS WHERE table_name = '{name}' AND owner = '{schema}'"
         )