SNOW-2306184: config refactor - variables merge

Author: sfc-gh-mraba

src/snowflake/cli/_plugins/dcm/manager.py

@@ -18,7 +18,6 @@
 import yaml
 from snowflake.cli._plugins.stage.manager import StageManager
 from snowflake.cli.api.artifacts.upload import sync_artifacts_with_stage
-from snowflake.cli.api.commands.utils import parse_key_value_variables
 from snowflake.cli.api.console.console import cli_console
 from snowflake.cli.api.constants import (
     DEFAULT_SIZE_LIMIT_MB,
@@ -102,8 +101,15 @@ def execute(
             if configuration:
                 query += f" CONFIGURATION {configuration}"
             if variables:
+                from snowflake.cli.api.commands.common import Variable
+                from snowflake.cli.api.config_ng import get_merged_variables
+
+                # Get merged variables from SnowSQL config and CLI -D parameters
+                merged_vars_dict = get_merged_variables(variables)
+                # Convert dict to List[Variable] for compatibility with parse_execute_variables
+                parsed_variables = [Variable(k, v) for k, v in merged_vars_dict.items()]
                 query += StageManager.parse_execute_variables(
-                    parse_key_value_variables(variables)
+                    parsed_variables
                 ).removeprefix(" using")
             stage_path = StagePath.from_stage_str(from_stage)
             query += f" FROM {stage_path.absolute_path()}"

src/snowflake/cli/_plugins/sql/commands.py

@@ -29,7 +29,6 @@
 )
 from snowflake.cli.api.commands.overrideable_parameter import OverrideableOption
 from snowflake.cli.api.commands.snow_typer import SnowTyperFactory
-from snowflake.cli.api.commands.utils import parse_key_value_variables
 from snowflake.cli.api.exceptions import CliArgumentError
 from snowflake.cli.api.output.types import (
     CommandResult,
@@ -136,9 +135,9 @@ def execute_sql(
     The command supports variable substitution that happens on client-side.
     """
 
-    data = {}
-    if data_override:
-        data = {v.key: v.value for v in parse_key_value_variables(data_override)}
+    from snowflake.cli.api.config_ng import get_merged_variables
+
+    data = get_merged_variables(data_override)
 
     template_syntax_config = _parse_template_syntax_config(enabled_templating)
 

src/snowflake/cli/_plugins/stage/manager.py

@@ -38,7 +38,6 @@
     OnErrorType,
     Variable,
 )
-from snowflake.cli.api.commands.utils import parse_key_value_variables
 from snowflake.cli.api.console import cli_console
 from snowflake.cli.api.constants import PYTHON_3_12
 from snowflake.cli.api.exceptions import CliError
@@ -608,7 +607,12 @@ def execute(
             filtered_file_list, key=lambda f: (path.dirname(f), path.basename(f))
         )
 
-        parsed_variables = parse_key_value_variables(variables)
+        from snowflake.cli.api.config_ng import get_merged_variables
+
+        # Get merged variables from SnowSQL config and CLI -D parameters
+        merged_vars_dict = get_merged_variables(variables)
+        # Convert dict back to List[Variable] for compatibility with existing methods
+        parsed_variables = [Variable(k, v) for k, v in merged_vars_dict.items()]
         sql_variables = self.parse_execute_variables(parsed_variables)
         python_variables = self._parse_python_variables(parsed_variables)
         results = []

src/snowflake/cli/api/config_ng/__init__.py

@@ -55,6 +55,8 @@
     ConnectionSpecificEnvironment,
     SnowSQLConfigFile,
     SnowSQLEnvironment,
+    SnowSQLSection,
+    get_merged_variables,
 )
 
 __all__ = [
@@ -69,6 +71,7 @@
     "explain_configuration",
     "export_resolution_history",
     "format_summary_for_display",
+    "get_merged_variables",
     "get_resolution_summary",
     "get_resolver",
     "is_resolution_logging_available",
@@ -80,6 +83,7 @@
     "show_resolution_chain",
     "SnowSQLConfigFile",
     "SnowSQLEnvironment",
+    "SnowSQLSection",
     "SourceType",
     "ValueSource",
 ]

src/snowflake/cli/api/config_ng/sources.py

@@ -31,13 +31,27 @@
 import configparser
 import logging
 import os
+from enum import Enum
 from pathlib import Path
-from typing import Any, Dict, Final, Optional
+from typing import Any, Dict, Final, List, Optional
 
 from snowflake.cli.api.config_ng.core import ConfigValue, SourceType, ValueSource
 
 log = logging.getLogger(__name__)
 
+
+class SnowSQLSection(Enum):
+    """
+    SnowSQL configuration file section names.
+
+    These sections can be present in SnowSQL INI config files.
+    """
+
+    CONNECTIONS = "connections"
+    VARIABLES = "variables"
+    OPTIONS = "options"
+
+
 # Try to import tomllib (Python 3.11+) or fall back to tomli
 try:
     import tomllib
@@ -142,9 +156,9 @@ def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
 
                 # Process all connection sections
                 for section in config.sections():
-                    if section.startswith("connections"):
+                    if section.startswith(SnowSQLSection.CONNECTIONS.value):
                         # Extract connection name
-                        if section == "connections":
+                        if section == SnowSQLSection.CONNECTIONS.value:
                             # This is default connection
                             connection_name = "default"
                         else:
@@ -170,6 +184,19 @@ def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
                                     raw_value=f"{param_key}={param_value}",  # Show original key in raw_value
                                 )
 
+                    elif section == SnowSQLSection.VARIABLES.value:
+                        # Process variables section (global, not connection-specific)
+                        section_data = dict(config[section])
+                        for var_key, var_value in section_data.items():
+                            full_key = f"variables.{var_key}"
+                            if key is None or full_key == key:
+                                merged_values[full_key] = ConfigValue(
+                                    key=full_key,
+                                    value=var_value,
+                                    source_name=self.source_name,
+                                    raw_value=f"{var_key}={var_value}",
+                                )
+
             except Exception as e:
                 log.debug("Failed to read SnowSQL config %s: %s", config_file, e)
 
@@ -755,3 +782,36 @@ def discover(self, key: Optional[str] = None) -> Dict[str, ConfigValue]:
     def supports_key(self, key: str) -> bool:
         """Check if key is present in CLI context with non-None value."""
         return key in self._cli_context and self._cli_context[key] is not None
+
+
+def get_merged_variables(cli_variables: Optional[List[str]] = None) -> Dict[str, str]:
+    """
+    Merge SnowSQL [variables] with CLI -D parameters.
+
+    Precedence: SnowSQL variables (lower) < -D parameters (higher)
+
+    Args:
+        cli_variables: List of "key=value" strings from -D parameters
+
+    Returns:
+        Dictionary of merged variables (key -> value)
+    """
+    from snowflake.cli.api.config_provider import get_config_provider_singleton
+
+    # Start with SnowSQL variables from config
+    provider = get_config_provider_singleton()
+    try:
+        snowsql_vars = provider.get_section(SnowSQLSection.VARIABLES.value)
+    except Exception:
+        # If variables section doesn't exist or provider not initialized, start with empty dict
+        snowsql_vars = {}
+
+    # Parse and overlay -D parameters (higher precedence)
+    if cli_variables:
+        from snowflake.cli.api.commands.utils import parse_key_value_variables
+
+        cli_vars_parsed = parse_key_value_variables(cli_variables)
+        for var in cli_vars_parsed:
+            snowsql_vars[var.key] = var.value
+
+    return snowsql_vars

src/snowflake/cli/api/config_provider.py

@@ -335,6 +335,15 @@ def get_section(self, *path) -> dict:
             connection_name = path[1]
             return self._get_connection_dict_internal(connection_name)
 
+        # For variables section, return all variables as flat dict
+        if len(path) == 1 and path[0] == "variables":
+            result = {}
+            for key, value in self._config_cache.items():
+                if key.startswith("variables."):
+                    var_name = key[len("variables.") :]
+                    result[var_name] = value
+            return result
+
         # For other sections, try to resolve with path prefix
         section_prefix = ".".join(path)
         result = {}