Merge branch 'main' of https://github.com/luisggc/sqlcompare

Author: luisggoncalves

sqlcompare/compare/comparator.py

@@ -260,13 +260,10 @@ def compare(
             except Exception:
                 pass
 
-            # Get column names from pre-existing tables
-            _, cols_prev = db.query(
-                f"SELECT * FROM {tables['previous']} WHERE 1=0", include_columns=True
-            )
-            _, cols_new = db.query(
-                f"SELECT * FROM {tables['new']} WHERE 1=0", include_columns=True
-            )
+            # Get column names from pre-existing tables using database-specific metadata
+            # This ensures we get the actual column case (important for Snowflake)
+            cols_prev = db.get_table_columns(tables['previous'])
+            cols_new = db.get_table_columns(tables['new'])
             self.cols_prev = cols_prev
             self.cols_new = cols_new
 
@@ -420,17 +417,15 @@ def compare(
         runs = load_test_runs()
 
         # Save metadata for later analysis
-        # For DuckDB connections (file-based tests), save "duckdb"
-        # For remote databases (Snowflake, etc.), save the connection name
-        conn_name = self.connection if isinstance(self.connection, str) else "duckdb"
-
+        # Save the connection as-is (can be None, connection ID, or URL)
+        # When None, it will resolve to the default connection on inspection
         run_data = {
             "tables": tables,
             "index_cols": list(self.index_cols),
             "cols_prev": self.cols_prev,
             "cols_new": self.cols_new,
             "common_cols": self.common_cols,
-            "conn": conn_name,
+            "conn": self.connection,
         }
 
         runs[diff_id] = run_data

sqlcompare/config.py

@@ -3,11 +3,12 @@
 import os
 from pathlib import Path
 from typing import Any
+import yaml
 
 
 def _config_dir() -> Path:
     return Path(
-        os.getenv("SQLCOMPARE_CONFIG_DIR", Path.home() / ".config" / "data-toolkit")
+        os.getenv("SQLCOMPARE_CONFIG_DIR", Path.home() / ".config" / "sqlcompare")
     )
 
 
@@ -23,8 +24,8 @@ def load_config() -> dict[str, Any]:
     }
 
 
-def _runs_file() -> Path:
-    return _config_dir() / "db_test_runs.yaml"
+def _runs_dir() -> Path:
+    return _config_dir() / "runs"
 
 
 def get_tests_folder() -> Path:
@@ -35,39 +36,32 @@ def get_tests_folder() -> Path:
 
 
 def load_test_runs() -> dict[str, Any]:
-    path = _runs_file()
-    if not path.exists():
+    runs_dir = _runs_dir()
+    if not runs_dir.exists():
         return {}
-    payload = _read_yaml(path)
-    if not payload:
-        return {}
-    if not isinstance(payload, dict):
-        raise ValueError("Expected mapping in db_test_runs.yaml")
-    return payload
+
+    runs = {}
+    for yaml_file in runs_dir.glob("*.yaml"):
+        run_id = yaml_file.stem  # filename without extension
+        payload = _read_yaml(yaml_file)
+        if payload:
+            runs[run_id] = payload
+
+    return runs
 
 
 def save_test_runs(runs: dict[str, Any]) -> None:
-    path = _runs_file()
-    path.parent.mkdir(parents=True, exist_ok=True)
-    _write_yaml(path, runs)
+    runs_dir = _runs_dir()
+    runs_dir.mkdir(parents=True, exist_ok=True)
 
+    for run_id, run_data in runs.items():
+        file_path = runs_dir / f"{run_id}.yaml"
+        _write_yaml(file_path, run_data)
 
-def _read_yaml(path: Path) -> dict[str, Any] | None:
-    try:
-        import yaml
-    except ImportError:
-        import json
 
-        return json.loads(path.read_text(encoding="utf-8"))
+def _read_yaml(path: Path) -> dict[str, Any] | None:
     return yaml.safe_load(path.read_text(encoding="utf-8"))
 
 
 def _write_yaml(path: Path, payload: dict[str, Any]) -> None:
-    try:
-        import yaml
-    except ImportError:
-        import json
-
-        path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
-        return
     path.write_text(yaml.safe_dump(payload, sort_keys=True), encoding="utf-8")

sqlcompare/db/connection.py

@@ -78,7 +78,7 @@ def __enter__(self) -> "DBConnection":
 
         return self
 
-    def __exit__(self, exc_type, _exc, _tb) -> None:
+    def __exit__(self, exc_type, _exc_val, _exc_tb) -> None:
         # Commit / rollback transaction
         try:
             if self._tx is not None:
@@ -196,12 +196,22 @@ def _run(self, sql: str, params: dict[str, Any] | None = None) -> Result:
             res = self.conn.execute(text(sql), params or {})
             return res
         except SQLAlchemyError as e:
+            # Extract just the database error message without the full SQL statement
+            error_msg = str(e)
+            # SQLAlchemy often includes the SQL in square brackets at the end
+            # Format: "error message [SQL: long query here]"
+            if "[SQL:" in error_msg:
+                # Extract just the part before [SQL:
+                db_error = error_msg.split("[SQL:")[0].strip()
+            else:
+                db_error = error_msg
+
             raise DBConnectionError(
-                "SQL execution failed.",
+                db_error,
                 conn_id=self.conn_id,
                 sql=sql,
                 original=e,
-            ) from e
+            ) from None  # Suppress the original exception chain to avoid showing SQL twice
         finally:
             self.last_elapsed_ms = int((time.perf_counter() - t0) * 1000)
 
@@ -220,6 +230,37 @@ def execute(self, sql: str, params: dict[str, Any] | None = None) -> ExecMeta:
             elapsed_ms=self.last_elapsed_ms or 0, rowcount=rowcount, columns=cols
         )
 
+    def get_table_columns(self, table_name: str) -> list[str]:
+        """
+        Get the actual column names from a table, using database-specific metadata queries
+        to ensure correct case sensitivity (especially important for Snowflake).
+
+        Args:
+            table_name: Fully qualified table name (e.g., "schema.table" or "db.schema.table")
+
+        Returns:
+            List of column names with their actual case as stored in the database
+        """
+        # Detect if we're using Snowflake by checking the dialect
+        try:
+            dialect_name = self._engine.dialect.name.lower() if self._engine else None
+        except Exception:
+            dialect_name = None
+
+        # For Snowflake, use DESCRIBE TABLE to get actual column names
+        if dialect_name == "snowflake":
+            try:
+                # DESCRIBE TABLE returns columns with their actual case
+                result = self.query(f"DESCRIBE TABLE {table_name}")
+                # First column is the column name
+                return [row[0] for row in result]
+            except Exception:
+                pass
+
+        # Fallback: use SELECT * WHERE 1=0 and get column names from result
+        _, columns = self.query(f"SELECT * FROM {table_name} WHERE 1=0", include_columns=True)
+        return columns
+
     def query(
         self,
         sql: str,

sqlcompare/db/constants.py

@@ -4,11 +4,11 @@
 from pathlib import Path
 
 lib_name = "sqlcompare"
-lib_config_folder = Path.home() / f".{lib_name}"
+lib_config_folder = Path.home() / ".config" / f"{lib_name}"
 
 LIBRARY_CONNECTIONS = [
     os.getenv(
-        "SQLCOMPARE_CONNECTIONS_FILE", str(lib_config_folder / "connections.yml")
+        "SQLCOMPARE_CONNECTIONS_FILE", str(lib_config_folder / "connections.yaml")
     ),
     os.getenv("DTK_CONNECTIONS_FILE", str(Path.home() / ".dtk" / "connections.yml")),
 ]

sqlcompare/db/exceptions.py

@@ -11,3 +11,17 @@ def __init__(
         self.conn_id = conn_id
         self.sql = sql
         self.original = original
+
+    def __str__(self) -> str:
+        # The main error message is already set in __init__, just add SQL context if helpful
+        msg = super().__str__()
+
+        # Optionally show a brief SQL snippet for context
+        if self.sql and len(self.sql) > 200:
+            # Only show snippet if SQL is long (short SQL is fine to show in full)
+            sql_lines = self.sql.strip().split('\n')
+            first_line = sql_lines[0] if sql_lines else self.sql
+            sql_snippet = first_line[:100] + "..." if len(first_line) > 100 else first_line
+            msg += f"\n\nSQL operation: {sql_snippet}"
+
+        return msg

sqlcompare/db/resolver.py

@@ -23,10 +23,27 @@ def normalize_connection_name(name: str) -> str:
     return s
 
 
+def _expand_env_vars(value: Any) -> Any:
+    """
+    Recursively expand environment variables in strings, dicts, and lists.
+
+    Expands ${VAR_NAME} and $VAR_NAME patterns in strings.
+    """
+    if isinstance(value, str):
+        return os.path.expandvars(value)
+    elif isinstance(value, dict):
+        return {k: _expand_env_vars(v) for k, v in value.items()}
+    elif isinstance(value, list):
+        return [_expand_env_vars(item) for item in value]
+    else:
+        return value
+
+
 def _load_connections_from_yaml(yaml_path: str) -> dict[str, dict[str, Any]]:
     """
     Load connection definitions from YAML file.
     Returns a dict mapping conn_id -> URL.create() parameters.
+    Expands environment variables in all string values.
     """
     expanded_path = Path(yaml_path).expanduser()
     if not expanded_path.exists():
@@ -35,7 +52,10 @@ def _load_connections_from_yaml(yaml_path: str) -> dict[str, dict[str, Any]]:
     try:
         with open(expanded_path, "r") as f:
             data = yaml.safe_load(f)
-            return data if isinstance(data, dict) else {}
+            if isinstance(data, dict):
+                # Expand environment variables in all values
+                return _expand_env_vars(data)
+            return {}
     except Exception:
         return {}
 

sqlcompare/query.py

@@ -37,7 +37,7 @@ def query(q: str, connection: str | None, output: str) -> None:
         if rows:
             from tabulate import tabulate
 
-            print(tabulate(rows, headers=columns))
+            print(tabulate(rows, headers=columns, tablefmt="pretty"))
             print(f"\nTotal rows: {len(rows)}")
         else:
             print("Query executed successfully. No rows returned.")

sqlcompare/utils/format.py

@@ -33,7 +33,7 @@ def format_table(
     """
     n_cols = len(columns)
     if n_cols == 0:
-        return tabulate([], headers=[], tablefmt=tablefmt, **tabulate_kwargs)
+        return tabulate([], headers=[], tablefmt="pretty", **tabulate_kwargs)
 
     # ---- column trimming ----
     use_col_ellipsis = n_cols > max_cols and max_cols >= 2

sqlcompare/utils/test_types/stats.py

@@ -6,7 +6,7 @@
 
 
 def _quote_ident(name: str) -> str:
-    return f'"{name.replace('"', '""')}"'
+    return '"' + name.replace('"', '""') + '"'
 
 
 def _collect_table_stats(
@@ -69,12 +69,8 @@ def compare_table_stats(table1: str, table2: str, connection: str | None) -> Non
             db.create_table_from_file(table1_name, table1)
             db.create_table_from_file(table2_name, table2)
 
-        _, cols_prev = db.query(
-            f"SELECT * FROM {table1_name} WHERE 1=0", include_columns=True
-        )
-        _, cols_new = db.query(
-            f"SELECT * FROM {table2_name} WHERE 1=0", include_columns=True
-        )
+        cols_prev = db.get_table_columns(table1_name)
+        cols_new = db.get_table_columns(table2_name)
 
         prev_map = {col.upper(): col for col in cols_prev}
         new_map = {col.upper(): col for col in cols_new}

tests/cli_helpers.py

@@ -29,3 +29,9 @@ def set_cli_env(
     monkeypatch.setenv("SQLCOMPARE_CONN_DEFAULT", connection_name)
     monkeypatch.setenv(f"SQLCOMPARE_CONN_{connection_name.upper()}", connection_url)
     monkeypatch.setenv("SQLCOMPARE_COMPARISON_SCHEMA", schema)
+
+
+def setup_duckdb_env(monkeypatch, config_dir: Path, schema: str = "sqlcompare") -> None:
+    """Setup environment for DuckDB tests without a specific connection."""
+    monkeypatch.setenv("SQLCOMPARE_CONFIG_DIR", str(config_dir))
+    monkeypatch.setenv("SQLCOMPARE_COMPARISON_SCHEMA", schema)