Feat: Add support for dot env variables

Author: themisvaltinos

docs/guides/configuration.md

@@ -98,7 +98,32 @@ All software runs within a system environment that stores information as "enviro
 
 SQLMesh can access environment variables during configuration, which enables approaches like storing passwords/secrets outside the configuration file and changing configuration parameters dynamically based on which user is running SQLMesh.
 
-You can use environment variables in two ways: specifying them in the configuration file or creating properly named variables to override configuration file values.
+You can specify environment variables in the configuration file or by storing them in a `.env` file.
+
+### .env files
+
+SQLMesh automatically loads environment variables from a `.env` file in your project directory. This provides a convenient way to manage environment variables without having to set them in your shell.
+
+Create a `.env` file in your project root with key-value pairs:
+
+```bash
+# .env file
+SNOWFLAKE_PW=my_secret_password
+S3_BUCKET=s3://my-data-bucket/warehouse
+DATABASE_URL=postgresql://user:pass@localhost/db
+
+# Override specific SQLMesh configuration values
+SQLMESH__DEFAULT_GATEWAY=production
+SQLMESH__MODEL_DEFAULTS__DIALECT=snowflake
+```
+
+See the [overrides](#overrides) section for a detailed explanation of how these are defined.
+
+The rest of the `.env` file variables can be used in your configuration files with `{{ env_var('VARIABLE_NAME') }}` syntax in YAML or accessed via `os.environ['VARIABLE_NAME']` in Python.
+
+**Important considerations:**
+- Add `.env` to your `.gitignore` file to avoid committing sensitive information
+- SQLMesh will only load the `.env` file if it exists in the project directory
 
 ### Configuration file
 

sqlmesh/core/config/loader.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 
 from pydantic import ValidationError
+from dotenv import load_dotenv
 from sqlglot.helper import ensure_list
 
 from sqlmesh.core import constants as c
@@ -35,6 +36,9 @@ def load_configs(
         for p in (glob.glob(str(path)) or [str(path)])
     ]
 
+    for path in absolute_paths:
+        load_dotenv(dotenv_path=path / ".env", override=True)
+
     if not isinstance(config, str):
         if type(config) != config_type:
             config = convert_config_type(config, config_type)

tests/core/test_config.py

@@ -24,6 +24,7 @@
     load_config_from_env,
     load_config_from_paths,
     load_config_from_python_module,
+    load_configs,
 )
 from sqlmesh.core.context import Context
 from sqlmesh.core.engine_adapter.athena import AthenaEngineAdapter
@@ -1132,3 +1133,106 @@ def test_environment_suffix_target_catalog(tmp_path: Path) -> None:
             Config,
             project_paths=[config_path],
         )
+
+
+def test_load_python_config_dot_env_vars(tmp_path_factory):
+    main_dir = tmp_path_factory.mktemp("python_config")
+    config_path = main_dir / "config.py"
+    with open(config_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """from sqlmesh.core.config import Config, DuckDBConnectionConfig, GatewayConfig, ModelDefaultsConfig
+config = Config(gateways={"duckdb_gateway": GatewayConfig(connection=DuckDBConnectionConfig())}, model_defaults=ModelDefaultsConfig(dialect=''))
+        """
+        )
+
+    # The environment variable value from the dot env file should be set
+    # SQLMESH__ variables override config fields directly if they follow the naming structure
+    dot_path = main_dir / ".env"
+    with open(dot_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """SQLMESH__GATEWAYS__DUCKDB_GATEWAY__STATE_CONNECTION__TYPE="bigquery"
+SQLMESH__GATEWAYS__DUCKDB_GATEWAY__STATE_CONNECTION__CHECK_IMPORT="false"
+SQLMESH__DEFAULT_GATEWAY="duckdb_gateway"
+        """
+        )
+
+    # Use mock.patch.dict to isolate environment variables between the tests
+    with mock.patch.dict(os.environ, {}, clear=True):
+        configs = load_configs(
+            "config",
+            Config,
+            paths=[main_dir],
+        )
+
+    assert next(iter(configs.values())) == Config(
+        gateways={
+            "duckdb_gateway": GatewayConfig(
+                connection=DuckDBConnectionConfig(),
+                state_connection=BigQueryConnectionConfig(check_import=False),
+            ),
+        },
+        model_defaults=ModelDefaultsConfig(dialect=""),
+        default_gateway="duckdb_gateway",
+    )
+
+
+def test_load_yaml_config_dot_env_vars(tmp_path_factory):
+    main_dir = tmp_path_factory.mktemp("yaml_config")
+    config_path = main_dir / "config.yaml"
+    with open(config_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """gateways:
+  duckdb_gateway:
+    connection:
+      type: duckdb
+      catalogs:
+        local: local.db
+        cloud_sales: {{ env_var('S3_BUCKET') }}
+      extensions:
+        - name: httpfs
+      secrets:
+        - type: "s3"
+          key_id: {{ env_var('S3_KEY') }}
+          secret: {{ env_var('S3_SECRET') }}
+model_defaults:
+  dialect: ""
+"""
+        )
+
+    # This test checks both using SQLMESH__ prefixed environment variables with underscores
+    # and setting a regular environment variable for use with env_var().
+    dot_path = main_dir / ".env"
+    with open(dot_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """S3_BUCKET="s3://metrics_bucket/sales.db"
+S3_KEY="S3_KEY_ID"
+S3_SECRET="XXX_S3_SECRET_XXX"
+SQLMESH__DEFAULT_GATEWAY="duckdb_gateway"
+SQLMESH__MODEL_DEFAULTS__DIALECT="athena"
+"""
+        )
+
+    # Use mock.patch.dict to isolate environment variables between the tests
+    with mock.patch.dict(os.environ, {}, clear=True):
+        configs = load_configs(
+            "config",
+            Config,
+            paths=[main_dir],
+        )
+
+    assert next(iter(configs.values())) == Config(
+        gateways={
+            "duckdb_gateway": GatewayConfig(
+                connection=DuckDBConnectionConfig(
+                    catalogs={
+                        "local": "local.db",
+                        "cloud_sales": "s3://metrics_bucket/sales.db",
+                    },
+                    extensions=[{"name": "httpfs"}],
+                    secrets=[{"type": "s3", "key_id": "S3_KEY_ID", "secret": "XXX_S3_SECRET_XXX"}],
+                ),
+            ),
+        },
+        default_gateway="duckdb_gateway",
+        model_defaults=ModelDefaultsConfig(dialect="athena"),
+    )