add ability to provide custom path

Author: themisvaltinos

docs/guides/configuration.md

@@ -121,9 +121,19 @@ See the [overrides](#overrides) section for a detailed explanation of how these
 
 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.
 
+#### Custom dot env file location and name
+
+By default, SQLMesh loads `.env` files from each project directory. Alternatively, you can export the `SQLMESH_DOTENV_PATH` environment variable to specify a custom path and persist it across commands:
+
+```bash
+export SQLMESH_DOTENV_PATH=/path/to/custom/.custom_env
+sqlmesh plan
+```
+
 **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
+- SQLMesh will only load the `.env` file if it exists in the project directory (unless a custom path is specified)
+- When using a custom path, that specific file takes precedence over any `.env` file in the project directory.
 
 ### Configuration file
 

pyproject.toml

@@ -79,6 +79,7 @@ dev = [
     "PyAthena[Pandas]",
     "PyGithub>=2.6.0",
     "pyperf",
+    "python-dotenv",
     "pyspark~=3.5.0",
     "pytest",
     "pytest-asyncio",

sqlmesh/cli/main.py

@@ -84,6 +84,12 @@ def _sqlmesh_version() -> str:
     type=str,
     help="The directory to write log files to.",
 )
+@click.option(
+    "--dotenv",
+    type=click.Path(exists=True, path_type=Path),
+    help="Path to a custom .env file to load environment variables.",
+    envvar="SQLMESH_DOTENV_PATH",
+)
 @click.pass_context
 @error_handler
 def cli(
@@ -95,6 +101,7 @@ def cli(
     debug: bool = False,
     log_to_stdout: bool = False,
     log_file_dir: t.Optional[str] = None,
+    dotenv: t.Optional[Path] = None,
 ) -> None:
     """SQLMesh command line tool."""
     if "--help" in sys.argv:
@@ -118,7 +125,7 @@ def cli(
     )
     configure_console(ignore_warnings=ignore_warnings)
 
-    configs = load_configs(config, Context.CONFIG_TYPE, paths)
+    configs = load_configs(config, Context.CONFIG_TYPE, paths, dotenv_path=dotenv)
     log_limit = list(configs.values())[0].log_limit
 
     remove_excess_logs(log_file_dir, log_limit)

sqlmesh/core/config/loader.py

@@ -26,6 +26,7 @@ def load_configs(
     config_type: t.Type[C],
     paths: t.Union[str | Path, t.Iterable[str | Path]],
     sqlmesh_path: t.Optional[Path] = None,
+    dotenv_path: t.Optional[Path] = None,
 ) -> t.Dict[Path, C]:
     sqlmesh_path = sqlmesh_path or c.SQLMESH_PATH
     config = config or "config"
@@ -36,8 +37,11 @@ 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 dotenv_path:
+        load_dotenv(dotenv_path=dotenv_path, override=True)
+    else:
+        for path in absolute_paths:
+            load_dotenv(dotenv_path=path / ".env", override=True)
 
     if not isinstance(config, str):
         if type(config) != config_type:

sqlmesh/magics.py

@@ -8,6 +8,7 @@
 from argparse import Namespace, SUPPRESS
 from collections import defaultdict
 from copy import deepcopy
+from pathlib import Path
 
 from hyperscript import h
 
@@ -166,6 +167,9 @@ def _shell(self) -> t.Any:
     @argument("--ignore-warnings", action="store_true", help="Ignore warnings.")
     @argument("--debug", action="store_true", help="Enable debug mode.")
     @argument("--log-file-dir", type=str, help="The directory to write the log file to.")
+    @argument(
+        "--dotenv", type=str, help="Path to a custom .env file to load environment variables from."
+    )
     @line_magic
     def context(self, line: str) -> None:
         """Sets the context in the user namespace."""
@@ -181,7 +185,10 @@ def context(self, line: str) -> None:
         )
         configure_console(ignore_warnings=args.ignore_warnings)
 
-        configs = load_configs(args.config, Context.CONFIG_TYPE, args.paths)
+        dotenv_path = Path(args.dotenv) if args.dotenv else None
+        configs = load_configs(
+            args.config, Context.CONFIG_TYPE, args.paths, dotenv_path=dotenv_path
+        )
         log_limit = list(configs.values())[0].log_limit
 
         remove_excess_logs(log_file_dir, log_limit)

tests/core/test_config.py

@@ -1236,3 +1236,60 @@ def test_load_yaml_config_dot_env_vars(tmp_path_factory):
         default_gateway="duckdb_gateway",
         model_defaults=ModelDefaultsConfig(dialect="athena"),
     )
+
+
+def test_load_yaml_config_custom_dotenv_path(tmp_path_factory):
+    main_dir = tmp_path_factory.mktemp("yaml_config_2")
+    config_path = main_dir / "config.yaml"
+    with open(config_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """gateways:
+  test_gateway:
+    connection:
+      type: duckdb
+      database: {{ env_var('DB_NAME') }}
+"""
+        )
+
+    # Create a custom dot env file in a different location
+    custom_env_dir = tmp_path_factory.mktemp("custom_env")
+    custom_env_path = custom_env_dir / ".my_env"
+    with open(custom_env_path, "w", encoding="utf-8") as fd:
+        fd.write(
+            """DB_NAME="custom_database.db"
+SQLMESH__DEFAULT_GATEWAY="test_gateway"
+SQLMESH__MODEL_DEFAULTS__DIALECT="postgres"
+"""
+        )
+
+    # Test that without custom dotenv path, env vars are not loaded
+    with mock.patch.dict(os.environ, {}, clear=True):
+        with pytest.raises(
+            ConfigError, match=r"Default model SQL dialect is a required configuratio*"
+        ):
+            load_configs(
+                "config",
+                Config,
+                paths=[main_dir],
+            )
+
+    # Test that with custom dotenv path, env vars are loaded correctly
+    with mock.patch.dict(os.environ, {}, clear=True):
+        configs = load_configs(
+            "config",
+            Config,
+            paths=[main_dir],
+            dotenv_path=custom_env_path,
+        )
+
+    assert next(iter(configs.values())) == Config(
+        gateways={
+            "test_gateway": GatewayConfig(
+                connection=DuckDBConnectionConfig(
+                    database="custom_database.db",
+                ),
+            ),
+        },
+        default_gateway="test_gateway",
+        model_defaults=ModelDefaultsConfig(dialect="postgres"),
+    )