From 9d674c7bf03eca2d24fb8a1da1de8988dfe30cc6 Mon Sep 17 00:00:00 2001 From: guy Date: Thu, 1 Oct 2026 16:39:24 +0000 Subject: [PATCH 1/9] CORE-1496: support --dbt-vars in every edr command that runs the internal dbt project --- elementary/clients/dbt/factory.py | 17 +++ elementary/config/config.py | 6 +- elementary/monitor/cli.py | 26 ++-- .../alerts/data_monitoring_alerts.py | 7 +- .../data_monitoring/data_monitoring.py | 13 +- elementary/monitor/debug.py | 10 +- elementary/operations/cli.py | 2 + .../operations/upload_source_freshness.py | 10 +- elementary/utils/cli_utils.py | 30 ++++ tests/unit/cli/__init__.py | 0 tests/unit/cli/test_dbt_vars_option.py | 132 ++++++++++++++++++ .../dbt_runner/test_internal_dbt_runner.py | 115 +++++++++++++++ 12 files changed, 321 insertions(+), 47 deletions(-) create mode 100644 tests/unit/cli/__init__.py create mode 100644 tests/unit/cli/test_dbt_vars_option.py create mode 100644 tests/unit/clients/dbt_runner/test_internal_dbt_runner.py diff --git a/elementary/clients/dbt/factory.py b/elementary/clients/dbt/factory.py index 8751a9ec3..022926d7f 100644 --- a/elementary/clients/dbt/factory.py +++ b/elementary/clients/dbt/factory.py @@ -11,6 +11,8 @@ is_dbt2_binary_available, ) from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner +from elementary.config.config import Config +from elementary.monitor.dbt_project_utils import CLI_DBT_PROJECT_PATH class RunnerMethod(Enum): @@ -51,6 +53,21 @@ def create_dbt_runner( ) +def create_internal_dbt_runner( + config: Config, + force_dbt_deps: bool = False, +) -> CommandLineDbtRunner: + return create_dbt_runner( + CLI_DBT_PROJECT_PATH, + config.profiles_dir, + config.profile_target, + env_vars=config.env_vars, + vars=config.dbt_vars, + run_deps_if_needed=config.run_dbt_deps_if_needed, + force_dbt_deps=force_dbt_deps, + ) + + def get_dbt_runner_method() -> RunnerMethod: runner_method = os.getenv("DBT_RUNNER_METHOD") if runner_method: diff --git a/elementary/config/config.py b/elementary/config/config.py index fbce626c5..82eb72d59 100644 --- a/elementary/config/config.py +++ b/elementary/config/config.py @@ -1,6 +1,6 @@ import os from pathlib import Path -from typing import Optional +from typing import Any, Dict, Optional import google.auth # type: ignore[import] from dateutil import tz @@ -77,6 +77,7 @@ def __init__( project_name: Optional[str] = None, quiet_logs: Optional[bool] = None, ssl_ca_bundle: Optional[str] = None, + dbt_vars: Optional[Dict[str, Any]] = None, ): self.config_dir = config_dir self.profiles_dir = profiles_dir @@ -90,6 +91,9 @@ def __init__( self.env_vars = dict() self.env_vars.update(self._parse_dbt_quoting_to_env_vars(dbt_quoting)) + # Vars supplied to every run of the internal dbt project + self.dbt_vars: Dict[str, Any] = dict(dbt_vars or {}) + config = self._load_configuration() self.target_dir = self._first_not_none( diff --git a/elementary/monitor/cli.py b/elementary/monitor/cli.py index 8df8d20e2..98729270e 100644 --- a/elementary/monitor/cli.py +++ b/elementary/monitor/cli.py @@ -15,9 +15,7 @@ from elementary.monitor.debug import Debug from elementary.tracking.anonymous_tracking import AnonymousCommandLineTracking from elementary.utils import bucket_path -from elementary.utils.ordered_yaml import OrderedYaml - -yaml = OrderedYaml() +from elementary.utils.cli_utils import dbt_vars_option class Command: @@ -43,6 +41,7 @@ def decorator(func): default=False, help="Disable sampling of data. Useful if your data contains PII.", )(func) + func = dbt_vars_option(func) func = click.option( "--dbt-quoting", "-dq", @@ -238,12 +237,6 @@ def get_cli_properties() -> dict: help="Force running a full refresh of all incremental models in the edr dbt package (usually this is not needed, " "see documentation to learn more).", ) -@click.option( - "--dbt-vars", - type=str, - default=None, - help="Specify raw YAML string of your dbt variables.", -) @click.option( "--test", type=bool, @@ -355,7 +348,6 @@ def monitor( fg="bright_red", ) slack_webhook = deprecated_slack_webhook - vars = yaml.loads(dbt_vars) if dbt_vars else None config = Config( config_dir=config_dir, profiles_dir=profiles_dir, @@ -376,6 +368,7 @@ def monitor( maximum_columns_in_alert_samples=maximum_columns_in_alert_samples, quiet_logs=quiet_logs, ssl_ca_bundle=ssl_ca_bundle, + dbt_vars=dbt_vars, ) anonymous_tracking = AnonymousCommandLineTracking(config) anonymous_tracking.set_env("use_select", bool(select)) @@ -410,9 +403,7 @@ def monitor( anonymous_tracking.track_cli_start( Command.MONITOR, get_cli_properties(), ctx.command.name ) - success = data_monitoring.run_alerts( - days_back, full_refresh_dbt_package, dbt_vars=vars - ) + success = data_monitoring.run_alerts(days_back, full_refresh_dbt_package) anonymous_tracking.track_cli_end( Command.MONITOR, data_monitoring.properties(), ctx.command.name ) @@ -471,6 +462,7 @@ def report( select, target_path, quiet_logs, + dbt_vars, ): """ Generate a local observability report of your warehouse. @@ -485,6 +477,7 @@ def report( dbt_quoting=dbt_quoting, env=env, quiet_logs=quiet_logs, + dbt_vars=dbt_vars, ) anonymous_tracking = AnonymousCommandLineTracking(config) anonymous_tracking.set_env("use_select", bool(select)) @@ -704,6 +697,7 @@ def send_report( target_path, quiet_logs, ssl_ca_bundle, + dbt_vars, ): """ Generate and send the report to an external platform. @@ -748,6 +742,7 @@ def send_report( project_name=project_name, quiet_logs=quiet_logs, ssl_ca_bundle=ssl_ca_bundle, + dbt_vars=dbt_vars, ) anonymous_tracking = AnonymousCommandLineTracking(config) anonymous_tracking.set_env("use_select", bool(select)) @@ -809,9 +804,10 @@ def send_report( help="Which directory to look in for the profiles.yml file. " "If not set, edr will look in HOME/.dbt/", ) +@dbt_vars_option @click.pass_context -def debug(ctx, profiles_dir): - config = Config(profiles_dir=profiles_dir) +def debug(ctx, profiles_dir, dbt_vars): + config = Config(profiles_dir=profiles_dir, dbt_vars=dbt_vars) anonymous_tracking = AnonymousCommandLineTracking(config) anonymous_tracking.track_cli_start(Command.DEBUG, None, ctx.command.name) success = Debug(config).run() diff --git a/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py b/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py index 0b0158f4d..f3f4fee43 100644 --- a/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py +++ b/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py @@ -1,7 +1,7 @@ import json from collections import defaultdict from datetime import datetime, timezone -from typing import DefaultDict, Dict, List, Optional, Union +from typing import Any, DefaultDict, Dict, List, Optional, Union from alive_progress import alive_bar @@ -146,10 +146,9 @@ def _populate_data( self, days_back: Optional[int] = None, dbt_full_refresh: bool = False, - dbt_vars: Optional[dict] = None, ) -> bool: logger.info("Running internal dbt run to populate alerts") - vars = dbt_vars or dict() + vars: Dict[str, Any] = {} if days_back: vars.update(days_back=days_back) success = self.internal_dbt_runner.run( @@ -426,14 +425,12 @@ def run_alerts( self, days_back: int, dbt_full_refresh: bool = False, - dbt_vars: Optional[dict] = None, ) -> bool: # Populate data if self.should_populate_data: popopulated_data_successfully = self._populate_data( days_back=days_back, dbt_full_refresh=dbt_full_refresh, - dbt_vars=dbt_vars, ) if not popopulated_data_successfully: self.success = False diff --git a/elementary/monitor/data_monitoring/data_monitoring.py b/elementary/monitor/data_monitoring/data_monitoring.py index dde9fd000..3c78433b8 100644 --- a/elementary/monitor/data_monitoring/data_monitoring.py +++ b/elementary/monitor/data_monitoring/data_monitoring.py @@ -3,9 +3,8 @@ from packaging import version -from elementary.clients.dbt.factory import create_dbt_runner +from elementary.clients.dbt.factory import create_internal_dbt_runner from elementary.config.config import Config -from elementary.monitor import dbt_project_utils from elementary.monitor.data_monitoring.schema import FiltersSchema, WarehouseInfo from elementary.tracking.anonymous_tracking import AnonymousTracking from elementary.tracking.tracking_interface import Tracking @@ -58,15 +57,9 @@ def __init__( self.selector_filter = selector_filter def _init_internal_dbt_runner(self): - internal_dbt_runner = create_dbt_runner( - dbt_project_utils.CLI_DBT_PROJECT_PATH, - self.config.profiles_dir, - self.config.profile_target, - env_vars=self.config.env_vars, - run_deps_if_needed=self.config.run_dbt_deps_if_needed, - force_dbt_deps=self.force_update_dbt_package, + return create_internal_dbt_runner( + self.config, force_dbt_deps=self.force_update_dbt_package ) - return internal_dbt_runner def properties(self): data_monitoring_properties = { diff --git a/elementary/monitor/debug.py b/elementary/monitor/debug.py index b8d8af0ff..52f9b23e6 100644 --- a/elementary/monitor/debug.py +++ b/elementary/monitor/debug.py @@ -1,9 +1,8 @@ import click -from elementary.clients.dbt.factory import create_dbt_runner +from elementary.clients.dbt.factory import create_internal_dbt_runner from elementary.config.config import Config from elementary.exceptions.exceptions import DbtCommandError -from elementary.monitor import dbt_project_utils class Debug: @@ -11,12 +10,7 @@ def __init__(self, config: Config): self.config = config def run(self) -> bool: - dbt_runner = create_dbt_runner( - dbt_project_utils.CLI_DBT_PROJECT_PATH, - self.config.profiles_dir, - self.config.profile_target, - run_deps_if_needed=self.config.run_dbt_deps_if_needed, - ) + dbt_runner = create_internal_dbt_runner(self.config) try: dbt_runner.run_operation("elementary_cli.test_conn", quiet=True) diff --git a/elementary/operations/cli.py b/elementary/operations/cli.py index f92eb99d1..d9c40d244 100644 --- a/elementary/operations/cli.py +++ b/elementary/operations/cli.py @@ -3,6 +3,7 @@ from elementary.config.config import Config from elementary.operations.upload_source_freshness import UploadSourceFreshnessOperation from elementary.tracking.anonymous_tracking import AnonymousCommandLineTracking +from elementary.utils.cli_utils import dbt_vars_option _MODULE_NAME = "run-operation" @@ -51,6 +52,7 @@ def run_operation(): default=25, help="Amount of rows to insert per insert statement.", ) +@dbt_vars_option @click.pass_context def upload_source_freshness(ctx, rows_per_insert: int, **conf): """ diff --git a/elementary/operations/upload_source_freshness.py b/elementary/operations/upload_source_freshness.py index 73f4ad2b4..9c941fadd 100644 --- a/elementary/operations/upload_source_freshness.py +++ b/elementary/operations/upload_source_freshness.py @@ -5,9 +5,8 @@ import click from alive_progress import alive_it -from elementary.clients.dbt.factory import create_dbt_runner +from elementary.clients.dbt.factory import create_internal_dbt_runner from elementary.config.config import Config -from elementary.monitor import dbt_project_utils from elementary.utils.ordered_yaml import OrderedYaml @@ -36,12 +35,7 @@ def get_sources_file_contents(self) -> dict: return json.loads(source_path.read_text()) def upload_results(self, results: dict, metadata: dict, rows_per_insert: int): - dbt_runner = create_dbt_runner( - dbt_project_utils.CLI_DBT_PROJECT_PATH, - self.config.profiles_dir, - self.config.profile_target, - run_deps_if_needed=self.config.run_dbt_deps_if_needed, - ) + dbt_runner = create_internal_dbt_runner(self.config) invocation_id = metadata.get("invocation_id") if not invocation_id: diff --git a/elementary/utils/cli_utils.py b/elementary/utils/cli_utils.py index 377a5831b..003cbc526 100644 --- a/elementary/utils/cli_utils.py +++ b/elementary/utils/cli_utils.py @@ -1,6 +1,10 @@ +from typing import Any, Dict, Optional + import click from click import ClickException +from elementary.utils.ordered_yaml import OrderedYaml + class RequiredIf(click.Option): def __init__(self, *args, **kwargs): @@ -28,3 +32,29 @@ def handle_parse_result(self, ctx, opts, args): self.prompt = None return super(RequiredIf, self).handle_parse_result(ctx, opts, args) + + +def _parse_dbt_vars( + ctx: click.Context, param: click.Parameter, value: Optional[str] +) -> Optional[Dict[str, Any]]: + if not value: + return None + try: + dbt_vars = OrderedYaml().loads(value) + except Exception as exc: + raise click.BadParameter(f"Invalid YAML: {exc}") from exc + if not isinstance(dbt_vars, dict): + raise click.BadParameter( + "Must be a YAML mapping, for example '{query_max_size: 1000000}'." + ) + return dbt_vars + + +dbt_vars_option = click.option( + "--dbt-vars", + type=str, + default=None, + callback=_parse_dbt_vars, + help="Specify raw YAML string of your dbt variables. " + "Applied to every run of the edr internal dbt project.", +) diff --git a/tests/unit/cli/__init__.py b/tests/unit/cli/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/cli/test_dbt_vars_option.py b/tests/unit/cli/test_dbt_vars_option.py new file mode 100644 index 000000000..b35e4035f --- /dev/null +++ b/tests/unit/cli/test_dbt_vars_option.py @@ -0,0 +1,132 @@ +from typing import List +from unittest import mock + +import pytest +from click.testing import CliRunner + +from elementary.config.config import Config +from elementary.monitor.cli import monitor +from elementary.operations.cli import run_operation + +DBT_VARS = "{query_max_size: 1000, nested: {key: value}}" +EXPECTED_DBT_VARS = {"query_max_size": 1000, "nested": {"key": "value"}} + + +@pytest.fixture(autouse=True) +def isolated_env(tmp_path, monkeypatch): + monkeypatch.setenv("DBT_LOG_PATH", str(tmp_path)) + monkeypatch.setattr(Config, "DEFAULT_CONFIG_DIR", str(tmp_path)) + monkeypatch.setattr(Config, "DEFAULT_TARGET_PATH", str(tmp_path)) + with mock.patch("elementary.monitor.cli.AnonymousCommandLineTracking"), mock.patch( + "elementary.operations.cli.AnonymousCommandLineTracking" + ): + yield + + +def _common_args(tmp_path) -> List[str]: + return ["--config-dir", str(tmp_path), "--target-path", str(tmp_path)] + + +def _invoke(cli, args: List[str]): + result = CliRunner().invoke(cli, args, catch_exceptions=False) + assert result.exit_code == 0, result.output + return result + + +@pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) +def test_monitor_dbt_vars(tmp_path, dbt_vars_args): + with mock.patch( + "elementary.monitor.cli.DataMonitoringAlerts" + ) as data_monitoring_alerts: + _invoke( + monitor, + [*_common_args(tmp_path), "--slack-webhook", "mock", *dbt_vars_args], + ) + config = data_monitoring_alerts.call_args.kwargs["config"] + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + data_monitoring_alerts.return_value.run_alerts.assert_called_once_with(1, False) + + +@pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) +def test_report_dbt_vars(tmp_path, dbt_vars_args): + with mock.patch("elementary.monitor.cli.SelectorFilter"), mock.patch( + "elementary.monitor.cli.DataMonitoringReport" + ) as data_monitoring_report: + data_monitoring_report.return_value.generate_report.return_value = ( + True, + None, + ) + _invoke(monitor, ["report", *_common_args(tmp_path), *dbt_vars_args]) + config = data_monitoring_report.call_args.kwargs["config"] + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + + +@pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) +def test_send_report_dbt_vars(tmp_path, dbt_vars_args): + with mock.patch("elementary.monitor.cli.SelectorFilter"), mock.patch( + "elementary.monitor.cli.DataMonitoringReport" + ) as data_monitoring_report: + _invoke( + monitor, + [ + "send-report", + *_common_args(tmp_path), + "--slack-token", + "mock", + "--slack-channel-name", + "mock", + *dbt_vars_args, + ], + ) + config = data_monitoring_report.call_args.kwargs["config"] + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + + +@pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) +def test_debug_dbt_vars(dbt_vars_args): + with mock.patch("elementary.monitor.cli.Debug") as debug: + _invoke(monitor, ["debug", *dbt_vars_args]) + config = debug.call_args.args[0] + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + + +@pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) +def test_upload_source_freshness_dbt_vars(tmp_path, dbt_vars_args): + with mock.patch( + "elementary.operations.cli.UploadSourceFreshnessOperation" + ) as operation: + _invoke( + run_operation, + [ + "upload-source-freshness", + "--target-path", + str(tmp_path), + *dbt_vars_args, + ], + ) + config = operation.call_args.args[0] + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + + +@pytest.mark.parametrize( + "args", + [ + ["report"], + ["send-report"], + ["debug"], + ], +) +@pytest.mark.parametrize("dbt_vars", ["{invalid", "not_a_mapping"]) +def test_invalid_dbt_vars(args, dbt_vars): + result = CliRunner().invoke(monitor, [*args, "--dbt-vars", dbt_vars]) + assert result.exit_code == 2 + assert "--dbt-vars" in result.output + + +@pytest.mark.parametrize("dbt_vars", ["{invalid", "not_a_mapping"]) +def test_invalid_dbt_vars_upload_source_freshness(dbt_vars): + result = CliRunner().invoke( + run_operation, ["upload-source-freshness", "--dbt-vars", dbt_vars] + ) + assert result.exit_code == 2 + assert "--dbt-vars" in result.output diff --git a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py new file mode 100644 index 000000000..83394feb1 --- /dev/null +++ b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py @@ -0,0 +1,115 @@ +import json +from unittest import mock + +import pytest + +from elementary.clients.dbt.factory import create_internal_dbt_runner +from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner +from elementary.config.config import Config +from elementary.monitor.data_monitoring.data_monitoring import DataMonitoring +from elementary.monitor.dbt_project_utils import CLI_DBT_PROJECT_PATH +from elementary.monitor.debug import Debug +from elementary.operations.upload_source_freshness import UploadSourceFreshnessOperation +from tests.mocks.data_monitoring.alerts.data_monitoring_alerts_mock import ( + DataMonitoringAlertsMock, +) + +DBT_VARS = {"query_max_size": 1000} + + +@pytest.fixture +def config(tmp_path, monkeypatch): + monkeypatch.setenv("DBT_LOG_PATH", str(tmp_path)) + return Config( + config_dir=str(tmp_path), + target_path=str(tmp_path), + profiles_dir="profiles_dir", + profile_target="target", + dbt_quoting="all", + run_dbt_deps_if_needed=False, + dbt_vars=DBT_VARS, + ) + + +@pytest.fixture +def mock_create_dbt_runner(): + with mock.patch( + "elementary.clients.dbt.factory.create_dbt_runner" + ) as create_dbt_runner: + yield create_dbt_runner + + +def _assert_internal_runner_created(mock_create_dbt_runner, config: Config, **kwargs): + mock_create_dbt_runner.assert_called_once_with( + CLI_DBT_PROJECT_PATH, + config.profiles_dir, + config.profile_target, + env_vars=config.env_vars, + vars=DBT_VARS, + run_deps_if_needed=False, + force_dbt_deps=kwargs.get("force_dbt_deps", False), + ) + + +def test_config_dbt_vars_default(tmp_path, monkeypatch): + monkeypatch.setenv("DBT_LOG_PATH", str(tmp_path)) + assert Config(config_dir=str(tmp_path), target_path=str(tmp_path)).dbt_vars == {} + + +@pytest.mark.parametrize("force_dbt_deps", [True, False]) +def test_create_internal_dbt_runner(config, mock_create_dbt_runner, force_dbt_deps): + runner = create_internal_dbt_runner(config, force_dbt_deps=force_dbt_deps) + assert runner is mock_create_dbt_runner.return_value + _assert_internal_runner_created( + mock_create_dbt_runner, config, force_dbt_deps=force_dbt_deps + ) + + +def test_data_monitoring_internal_runner_vars(config, mock_create_dbt_runner): + data_monitoring = DataMonitoring.__new__(DataMonitoring) + data_monitoring.config = config + data_monitoring.force_update_dbt_package = True + data_monitoring._init_internal_dbt_runner() + _assert_internal_runner_created(mock_create_dbt_runner, config, force_dbt_deps=True) + + +def test_debug_internal_runner_vars(config, mock_create_dbt_runner): + assert Debug(config).run() + _assert_internal_runner_created(mock_create_dbt_runner, config) + + +def test_upload_source_freshness_internal_runner_vars(config, mock_create_dbt_runner): + UploadSourceFreshnessOperation(config).upload_results( + results=[], metadata={"invocation_id": "id"}, rows_per_insert=10 + ) + _assert_internal_runner_created(mock_create_dbt_runner, config) + + +@mock.patch("subprocess.run") +def test_runner_vars_merged_with_per_call_vars(mock_subprocess_run): + runner = SubprocessDbtRunner( + project_dir="proj_dir", + profiles_dir="prof_dir", + vars={"query_max_size": 1000, "days_back": 30}, + run_deps_if_needed=False, + ) + runner.run(select="model", vars={"days_back": 7}) + args = mock_subprocess_run.call_args[0][0] + assert json.loads(args[args.index("--vars") + 1]) == { + "query_max_size": 1000, + "days_back": 7, + } + + +@pytest.mark.parametrize("days_back,expected_vars", [(3, {"days_back": 3}), (0, {})]) +def test_populate_alerts_per_call_vars(days_back, expected_vars): + data_monitoring_alerts = DataMonitoringAlertsMock() + with mock.patch.object( + data_monitoring_alerts.internal_dbt_runner, "run", return_value=True + ) as run: + assert data_monitoring_alerts._populate_data(days_back=days_back) + run.assert_called_once_with( + select="elementary_cli.alerts.alerts_v2", + full_refresh=False, + vars=expected_vars, + ) From 3454fd525a64d1a86d42d703fa697262850db5df Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 12:41:14 +0000 Subject: [PATCH 2/9] Address review: dbt_vars default in Config signature, inline days_back vars, clearer --dbt-vars error/help --- elementary/config/config.py | 5 ++--- .../data_monitoring/alerts/data_monitoring_alerts.py | 7 ++----- elementary/utils/cli_utils.py | 9 ++++----- .../unit/clients/dbt_runner/test_internal_dbt_runner.py | 8 +++++--- 4 files changed, 13 insertions(+), 16 deletions(-) diff --git a/elementary/config/config.py b/elementary/config/config.py index 82eb72d59..20919e72b 100644 --- a/elementary/config/config.py +++ b/elementary/config/config.py @@ -77,7 +77,7 @@ def __init__( project_name: Optional[str] = None, quiet_logs: Optional[bool] = None, ssl_ca_bundle: Optional[str] = None, - dbt_vars: Optional[Dict[str, Any]] = None, + dbt_vars: Dict[str, Any] = {}, ): self.config_dir = config_dir self.profiles_dir = profiles_dir @@ -91,8 +91,7 @@ def __init__( self.env_vars = dict() self.env_vars.update(self._parse_dbt_quoting_to_env_vars(dbt_quoting)) - # Vars supplied to every run of the internal dbt project - self.dbt_vars: Dict[str, Any] = dict(dbt_vars or {}) + self.dbt_vars = dict(dbt_vars) config = self._load_configuration() diff --git a/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py b/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py index f3f4fee43..9d7c3d500 100644 --- a/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py +++ b/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py @@ -1,7 +1,7 @@ import json from collections import defaultdict from datetime import datetime, timezone -from typing import Any, DefaultDict, Dict, List, Optional, Union +from typing import DefaultDict, Dict, List, Optional, Union from alive_progress import alive_bar @@ -148,13 +148,10 @@ def _populate_data( dbt_full_refresh: bool = False, ) -> bool: logger.info("Running internal dbt run to populate alerts") - vars: Dict[str, Any] = {} - if days_back: - vars.update(days_back=days_back) success = self.internal_dbt_runner.run( select="elementary_cli.alerts.alerts_v2", full_refresh=dbt_full_refresh, - vars=vars, + vars={"days_back": days_back} if days_back else None, ) self.execution_properties["alerts_populate_success"] = success if not success: diff --git a/elementary/utils/cli_utils.py b/elementary/utils/cli_utils.py index 003cbc526..59a2a3e2b 100644 --- a/elementary/utils/cli_utils.py +++ b/elementary/utils/cli_utils.py @@ -36,16 +36,16 @@ def handle_parse_result(self, ctx, opts, args): def _parse_dbt_vars( ctx: click.Context, param: click.Parameter, value: Optional[str] -) -> Optional[Dict[str, Any]]: +) -> Dict[str, Any]: if not value: - return None + return {} try: dbt_vars = OrderedYaml().loads(value) except Exception as exc: raise click.BadParameter(f"Invalid YAML: {exc}") from exc if not isinstance(dbt_vars, dict): raise click.BadParameter( - "Must be a YAML mapping, for example '{query_max_size: 1000000}'." + "dbt vars must be a key-value mapping, for example '{query_max_size: 1000000}'." ) return dbt_vars @@ -55,6 +55,5 @@ def _parse_dbt_vars( type=str, default=None, callback=_parse_dbt_vars, - help="Specify raw YAML string of your dbt variables. " - "Applied to every run of the edr internal dbt project.", + help="Specify raw YAML string of your dbt variables.", ) diff --git a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py index 83394feb1..de4dc0c5b 100644 --- a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py +++ b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py @@ -39,7 +39,9 @@ def mock_create_dbt_runner(): yield create_dbt_runner -def _assert_internal_runner_created(mock_create_dbt_runner, config: Config, **kwargs): +def _assert_internal_runner_created( + mock_create_dbt_runner, config: Config, force_dbt_deps: bool = False +): mock_create_dbt_runner.assert_called_once_with( CLI_DBT_PROJECT_PATH, config.profiles_dir, @@ -47,7 +49,7 @@ def _assert_internal_runner_created(mock_create_dbt_runner, config: Config, **kw env_vars=config.env_vars, vars=DBT_VARS, run_deps_if_needed=False, - force_dbt_deps=kwargs.get("force_dbt_deps", False), + force_dbt_deps=force_dbt_deps, ) @@ -101,7 +103,7 @@ def test_runner_vars_merged_with_per_call_vars(mock_subprocess_run): } -@pytest.mark.parametrize("days_back,expected_vars", [(3, {"days_back": 3}), (0, {})]) +@pytest.mark.parametrize("days_back,expected_vars", [(3, {"days_back": 3}), (0, None)]) def test_populate_alerts_per_call_vars(days_back, expected_vars): data_monitoring_alerts = DataMonitoringAlertsMock() with mock.patch.object( From f09a2c2255c62aa8e3b6fafc80f6e44fc8b5aa05 Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 12:52:36 +0000 Subject: [PATCH 3/9] Keep dbt_vars as None when --dbt-vars is omitted --- elementary/config/config.py | 4 ++-- elementary/utils/cli_utils.py | 4 ++-- tests/unit/cli/test_dbt_vars_option.py | 10 +++++----- .../clients/dbt_runner/test_internal_dbt_runner.py | 2 +- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/elementary/config/config.py b/elementary/config/config.py index 20919e72b..49dfeaabe 100644 --- a/elementary/config/config.py +++ b/elementary/config/config.py @@ -77,7 +77,7 @@ def __init__( project_name: Optional[str] = None, quiet_logs: Optional[bool] = None, ssl_ca_bundle: Optional[str] = None, - dbt_vars: Dict[str, Any] = {}, + dbt_vars: Optional[Dict[str, Any]] = None, ): self.config_dir = config_dir self.profiles_dir = profiles_dir @@ -91,7 +91,7 @@ def __init__( self.env_vars = dict() self.env_vars.update(self._parse_dbt_quoting_to_env_vars(dbt_quoting)) - self.dbt_vars = dict(dbt_vars) + self.dbt_vars = dbt_vars config = self._load_configuration() diff --git a/elementary/utils/cli_utils.py b/elementary/utils/cli_utils.py index 59a2a3e2b..57cd1a684 100644 --- a/elementary/utils/cli_utils.py +++ b/elementary/utils/cli_utils.py @@ -36,9 +36,9 @@ def handle_parse_result(self, ctx, opts, args): def _parse_dbt_vars( ctx: click.Context, param: click.Parameter, value: Optional[str] -) -> Dict[str, Any]: +) -> Optional[Dict[str, Any]]: if not value: - return {} + return None try: dbt_vars = OrderedYaml().loads(value) except Exception as exc: diff --git a/tests/unit/cli/test_dbt_vars_option.py b/tests/unit/cli/test_dbt_vars_option.py index b35e4035f..126f32e2e 100644 --- a/tests/unit/cli/test_dbt_vars_option.py +++ b/tests/unit/cli/test_dbt_vars_option.py @@ -43,7 +43,7 @@ def test_monitor_dbt_vars(tmp_path, dbt_vars_args): [*_common_args(tmp_path), "--slack-webhook", "mock", *dbt_vars_args], ) config = data_monitoring_alerts.call_args.kwargs["config"] - assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else None) data_monitoring_alerts.return_value.run_alerts.assert_called_once_with(1, False) @@ -58,7 +58,7 @@ def test_report_dbt_vars(tmp_path, dbt_vars_args): ) _invoke(monitor, ["report", *_common_args(tmp_path), *dbt_vars_args]) config = data_monitoring_report.call_args.kwargs["config"] - assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else None) @pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) @@ -79,7 +79,7 @@ def test_send_report_dbt_vars(tmp_path, dbt_vars_args): ], ) config = data_monitoring_report.call_args.kwargs["config"] - assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else None) @pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) @@ -87,7 +87,7 @@ def test_debug_dbt_vars(dbt_vars_args): with mock.patch("elementary.monitor.cli.Debug") as debug: _invoke(monitor, ["debug", *dbt_vars_args]) config = debug.call_args.args[0] - assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else None) @pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) @@ -105,7 +105,7 @@ def test_upload_source_freshness_dbt_vars(tmp_path, dbt_vars_args): ], ) config = operation.call_args.args[0] - assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else {}) + assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else None) @pytest.mark.parametrize( diff --git a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py index de4dc0c5b..118e76d8b 100644 --- a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py +++ b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py @@ -55,7 +55,7 @@ def _assert_internal_runner_created( def test_config_dbt_vars_default(tmp_path, monkeypatch): monkeypatch.setenv("DBT_LOG_PATH", str(tmp_path)) - assert Config(config_dir=str(tmp_path), target_path=str(tmp_path)).dbt_vars == {} + assert Config(config_dir=str(tmp_path), target_path=str(tmp_path)).dbt_vars is None @pytest.mark.parametrize("force_dbt_deps", [True, False]) From 51d8029063b035134fb45051182f8a71907fd354 Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 12:58:35 +0000 Subject: [PATCH 4/9] Pass days_back to alerts_v2 through the runner-level dbt vars --- elementary/monitor/cli.py | 4 +++- .../alerts/data_monitoring_alerts.py | 8 +------- tests/unit/cli/test_dbt_vars_option.py | 20 ++++++++++++++----- .../dbt_runner/test_internal_dbt_runner.py | 6 ++---- 4 files changed, 21 insertions(+), 17 deletions(-) diff --git a/elementary/monitor/cli.py b/elementary/monitor/cli.py index 98729270e..911cc0112 100644 --- a/elementary/monitor/cli.py +++ b/elementary/monitor/cli.py @@ -368,7 +368,9 @@ def monitor( maximum_columns_in_alert_samples=maximum_columns_in_alert_samples, quiet_logs=quiet_logs, ssl_ca_bundle=ssl_ca_bundle, - dbt_vars=dbt_vars, + dbt_vars={**(dbt_vars or {}), "days_back": days_back} + if days_back + else dbt_vars, ) anonymous_tracking = AnonymousCommandLineTracking(config) anonymous_tracking.set_env("use_select", bool(select)) diff --git a/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py b/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py index 9d7c3d500..afd355095 100644 --- a/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py +++ b/elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py @@ -142,16 +142,11 @@ def _get_integration_client( tracking=self.tracking, ) - def _populate_data( - self, - days_back: Optional[int] = None, - dbt_full_refresh: bool = False, - ) -> bool: + def _populate_data(self, dbt_full_refresh: bool = False) -> bool: logger.info("Running internal dbt run to populate alerts") success = self.internal_dbt_runner.run( select="elementary_cli.alerts.alerts_v2", full_refresh=dbt_full_refresh, - vars={"days_back": days_back} if days_back else None, ) self.execution_properties["alerts_populate_success"] = success if not success: @@ -426,7 +421,6 @@ def run_alerts( # Populate data if self.should_populate_data: popopulated_data_successfully = self._populate_data( - days_back=days_back, dbt_full_refresh=dbt_full_refresh, ) if not popopulated_data_successfully: diff --git a/tests/unit/cli/test_dbt_vars_option.py b/tests/unit/cli/test_dbt_vars_option.py index 126f32e2e..21112d10d 100644 --- a/tests/unit/cli/test_dbt_vars_option.py +++ b/tests/unit/cli/test_dbt_vars_option.py @@ -33,18 +33,28 @@ def _invoke(cli, args: List[str]): return result -@pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) -def test_monitor_dbt_vars(tmp_path, dbt_vars_args): +@pytest.mark.parametrize( + "extra_args,expected_dbt_vars", + [ + ([], {"days_back": 1}), + (["--dbt-vars", DBT_VARS], {**EXPECTED_DBT_VARS, "days_back": 1}), + (["--days-back", "3"], {"days_back": 3}), + (["--dbt-vars", "{days_back: 10}", "--days-back", "3"], {"days_back": 3}), + ], +) +def test_monitor_dbt_vars(tmp_path, extra_args, expected_dbt_vars): with mock.patch( "elementary.monitor.cli.DataMonitoringAlerts" ) as data_monitoring_alerts: _invoke( monitor, - [*_common_args(tmp_path), "--slack-webhook", "mock", *dbt_vars_args], + [*_common_args(tmp_path), "--slack-webhook", "mock", *extra_args], ) config = data_monitoring_alerts.call_args.kwargs["config"] - assert config.dbt_vars == (EXPECTED_DBT_VARS if dbt_vars_args else None) - data_monitoring_alerts.return_value.run_alerts.assert_called_once_with(1, False) + assert config.dbt_vars == expected_dbt_vars + data_monitoring_alerts.return_value.run_alerts.assert_called_once_with( + expected_dbt_vars["days_back"], False + ) @pytest.mark.parametrize("dbt_vars_args", [[], ["--dbt-vars", DBT_VARS]]) diff --git a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py index 118e76d8b..4612dd963 100644 --- a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py +++ b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py @@ -103,15 +103,13 @@ def test_runner_vars_merged_with_per_call_vars(mock_subprocess_run): } -@pytest.mark.parametrize("days_back,expected_vars", [(3, {"days_back": 3}), (0, None)]) -def test_populate_alerts_per_call_vars(days_back, expected_vars): +def test_populate_alerts_uses_runner_vars(): data_monitoring_alerts = DataMonitoringAlertsMock() with mock.patch.object( data_monitoring_alerts.internal_dbt_runner, "run", return_value=True ) as run: - assert data_monitoring_alerts._populate_data(days_back=days_back) + assert data_monitoring_alerts._populate_data() run.assert_called_once_with( select="elementary_cli.alerts.alerts_v2", full_refresh=False, - vars=expected_vars, ) From 94c84690098de7fde1ce5cba148ba704655f07dd Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 13:01:34 +0000 Subject: [PATCH 5/9] Replace create_internal_dbt_runner with create_dbt_runner_from_config --- elementary/clients/dbt/factory.py | 6 +++--- elementary/monitor/data_monitoring/data_monitoring.py | 9 ++++++--- elementary/monitor/debug.py | 5 +++-- elementary/operations/upload_source_freshness.py | 5 +++-- .../unit/clients/dbt_runner/test_internal_dbt_runner.py | 8 +++++--- 5 files changed, 20 insertions(+), 13 deletions(-) diff --git a/elementary/clients/dbt/factory.py b/elementary/clients/dbt/factory.py index 022926d7f..4b1159804 100644 --- a/elementary/clients/dbt/factory.py +++ b/elementary/clients/dbt/factory.py @@ -12,7 +12,6 @@ ) from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner from elementary.config.config import Config -from elementary.monitor.dbt_project_utils import CLI_DBT_PROJECT_PATH class RunnerMethod(Enum): @@ -53,12 +52,13 @@ def create_dbt_runner( ) -def create_internal_dbt_runner( +def create_dbt_runner_from_config( config: Config, + project_dir: str, force_dbt_deps: bool = False, ) -> CommandLineDbtRunner: return create_dbt_runner( - CLI_DBT_PROJECT_PATH, + project_dir, config.profiles_dir, config.profile_target, env_vars=config.env_vars, diff --git a/elementary/monitor/data_monitoring/data_monitoring.py b/elementary/monitor/data_monitoring/data_monitoring.py index 3c78433b8..56e8e0d09 100644 --- a/elementary/monitor/data_monitoring/data_monitoring.py +++ b/elementary/monitor/data_monitoring/data_monitoring.py @@ -3,9 +3,10 @@ from packaging import version -from elementary.clients.dbt.factory import create_internal_dbt_runner +from elementary.clients.dbt.factory import create_dbt_runner_from_config from elementary.config.config import Config from elementary.monitor.data_monitoring.schema import FiltersSchema, WarehouseInfo +from elementary.monitor.dbt_project_utils import CLI_DBT_PROJECT_PATH from elementary.tracking.anonymous_tracking import AnonymousTracking from elementary.tracking.tracking_interface import Tracking from elementary.utils import package @@ -57,8 +58,10 @@ def __init__( self.selector_filter = selector_filter def _init_internal_dbt_runner(self): - return create_internal_dbt_runner( - self.config, force_dbt_deps=self.force_update_dbt_package + return create_dbt_runner_from_config( + self.config, + CLI_DBT_PROJECT_PATH, + force_dbt_deps=self.force_update_dbt_package, ) def properties(self): diff --git a/elementary/monitor/debug.py b/elementary/monitor/debug.py index 52f9b23e6..9e7d7d64e 100644 --- a/elementary/monitor/debug.py +++ b/elementary/monitor/debug.py @@ -1,8 +1,9 @@ import click -from elementary.clients.dbt.factory import create_internal_dbt_runner +from elementary.clients.dbt.factory import create_dbt_runner_from_config from elementary.config.config import Config from elementary.exceptions.exceptions import DbtCommandError +from elementary.monitor.dbt_project_utils import CLI_DBT_PROJECT_PATH class Debug: @@ -10,7 +11,7 @@ def __init__(self, config: Config): self.config = config def run(self) -> bool: - dbt_runner = create_internal_dbt_runner(self.config) + dbt_runner = create_dbt_runner_from_config(self.config, CLI_DBT_PROJECT_PATH) try: dbt_runner.run_operation("elementary_cli.test_conn", quiet=True) diff --git a/elementary/operations/upload_source_freshness.py b/elementary/operations/upload_source_freshness.py index 9c941fadd..7b9338b29 100644 --- a/elementary/operations/upload_source_freshness.py +++ b/elementary/operations/upload_source_freshness.py @@ -5,8 +5,9 @@ import click from alive_progress import alive_it -from elementary.clients.dbt.factory import create_internal_dbt_runner +from elementary.clients.dbt.factory import create_dbt_runner_from_config from elementary.config.config import Config +from elementary.monitor.dbt_project_utils import CLI_DBT_PROJECT_PATH from elementary.utils.ordered_yaml import OrderedYaml @@ -35,7 +36,7 @@ def get_sources_file_contents(self) -> dict: return json.loads(source_path.read_text()) def upload_results(self, results: dict, metadata: dict, rows_per_insert: int): - dbt_runner = create_internal_dbt_runner(self.config) + dbt_runner = create_dbt_runner_from_config(self.config, CLI_DBT_PROJECT_PATH) invocation_id = metadata.get("invocation_id") if not invocation_id: diff --git a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py index 4612dd963..7810da7a7 100644 --- a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py +++ b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py @@ -3,7 +3,7 @@ import pytest -from elementary.clients.dbt.factory import create_internal_dbt_runner +from elementary.clients.dbt.factory import create_dbt_runner_from_config from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner from elementary.config.config import Config from elementary.monitor.data_monitoring.data_monitoring import DataMonitoring @@ -59,8 +59,10 @@ def test_config_dbt_vars_default(tmp_path, monkeypatch): @pytest.mark.parametrize("force_dbt_deps", [True, False]) -def test_create_internal_dbt_runner(config, mock_create_dbt_runner, force_dbt_deps): - runner = create_internal_dbt_runner(config, force_dbt_deps=force_dbt_deps) +def test_create_dbt_runner_from_config(config, mock_create_dbt_runner, force_dbt_deps): + runner = create_dbt_runner_from_config( + config, CLI_DBT_PROJECT_PATH, force_dbt_deps=force_dbt_deps + ) assert runner is mock_create_dbt_runner.return_value _assert_internal_runner_created( mock_create_dbt_runner, config, force_dbt_deps=force_dbt_deps From d834d40192884b9be2806a0bbd7d72beeec803f9 Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 13:24:04 +0000 Subject: [PATCH 6/9] Inject days_back into dbt_vars in Config; drop runner-wiring tests --- elementary/config/config.py | 5 +- elementary/monitor/cli.py | 5 +- .../data_monitoring/data_monitoring.py | 3 +- .../dbt_runner/test_internal_dbt_runner.py | 117 ------------------ 4 files changed, 8 insertions(+), 122 deletions(-) delete mode 100644 tests/unit/clients/dbt_runner/test_internal_dbt_runner.py diff --git a/elementary/config/config.py b/elementary/config/config.py index 49dfeaabe..0d7c45083 100644 --- a/elementary/config/config.py +++ b/elementary/config/config.py @@ -78,6 +78,7 @@ def __init__( quiet_logs: Optional[bool] = None, ssl_ca_bundle: Optional[str] = None, dbt_vars: Optional[Dict[str, Any]] = None, + days_back: Optional[int] = None, ): self.config_dir = config_dir self.profiles_dir = profiles_dir @@ -91,7 +92,9 @@ def __init__( self.env_vars = dict() self.env_vars.update(self._parse_dbt_quoting_to_env_vars(dbt_quoting)) - self.dbt_vars = dbt_vars + self.dbt_vars = ( + {**(dbt_vars or {}), "days_back": days_back} if days_back else dbt_vars + ) config = self._load_configuration() diff --git a/elementary/monitor/cli.py b/elementary/monitor/cli.py index 911cc0112..f1c4cbbfd 100644 --- a/elementary/monitor/cli.py +++ b/elementary/monitor/cli.py @@ -368,9 +368,8 @@ def monitor( maximum_columns_in_alert_samples=maximum_columns_in_alert_samples, quiet_logs=quiet_logs, ssl_ca_bundle=ssl_ca_bundle, - dbt_vars={**(dbt_vars or {}), "days_back": days_back} - if days_back - else dbt_vars, + dbt_vars=dbt_vars, + days_back=days_back, ) anonymous_tracking = AnonymousCommandLineTracking(config) anonymous_tracking.set_env("use_select", bool(select)) diff --git a/elementary/monitor/data_monitoring/data_monitoring.py b/elementary/monitor/data_monitoring/data_monitoring.py index 56e8e0d09..ca501d51f 100644 --- a/elementary/monitor/data_monitoring/data_monitoring.py +++ b/elementary/monitor/data_monitoring/data_monitoring.py @@ -58,11 +58,12 @@ def __init__( self.selector_filter = selector_filter def _init_internal_dbt_runner(self): - return create_dbt_runner_from_config( + internal_dbt_runner = create_dbt_runner_from_config( self.config, CLI_DBT_PROJECT_PATH, force_dbt_deps=self.force_update_dbt_package, ) + return internal_dbt_runner def properties(self): data_monitoring_properties = { diff --git a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py b/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py deleted file mode 100644 index 7810da7a7..000000000 --- a/tests/unit/clients/dbt_runner/test_internal_dbt_runner.py +++ /dev/null @@ -1,117 +0,0 @@ -import json -from unittest import mock - -import pytest - -from elementary.clients.dbt.factory import create_dbt_runner_from_config -from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner -from elementary.config.config import Config -from elementary.monitor.data_monitoring.data_monitoring import DataMonitoring -from elementary.monitor.dbt_project_utils import CLI_DBT_PROJECT_PATH -from elementary.monitor.debug import Debug -from elementary.operations.upload_source_freshness import UploadSourceFreshnessOperation -from tests.mocks.data_monitoring.alerts.data_monitoring_alerts_mock import ( - DataMonitoringAlertsMock, -) - -DBT_VARS = {"query_max_size": 1000} - - -@pytest.fixture -def config(tmp_path, monkeypatch): - monkeypatch.setenv("DBT_LOG_PATH", str(tmp_path)) - return Config( - config_dir=str(tmp_path), - target_path=str(tmp_path), - profiles_dir="profiles_dir", - profile_target="target", - dbt_quoting="all", - run_dbt_deps_if_needed=False, - dbt_vars=DBT_VARS, - ) - - -@pytest.fixture -def mock_create_dbt_runner(): - with mock.patch( - "elementary.clients.dbt.factory.create_dbt_runner" - ) as create_dbt_runner: - yield create_dbt_runner - - -def _assert_internal_runner_created( - mock_create_dbt_runner, config: Config, force_dbt_deps: bool = False -): - mock_create_dbt_runner.assert_called_once_with( - CLI_DBT_PROJECT_PATH, - config.profiles_dir, - config.profile_target, - env_vars=config.env_vars, - vars=DBT_VARS, - run_deps_if_needed=False, - force_dbt_deps=force_dbt_deps, - ) - - -def test_config_dbt_vars_default(tmp_path, monkeypatch): - monkeypatch.setenv("DBT_LOG_PATH", str(tmp_path)) - assert Config(config_dir=str(tmp_path), target_path=str(tmp_path)).dbt_vars is None - - -@pytest.mark.parametrize("force_dbt_deps", [True, False]) -def test_create_dbt_runner_from_config(config, mock_create_dbt_runner, force_dbt_deps): - runner = create_dbt_runner_from_config( - config, CLI_DBT_PROJECT_PATH, force_dbt_deps=force_dbt_deps - ) - assert runner is mock_create_dbt_runner.return_value - _assert_internal_runner_created( - mock_create_dbt_runner, config, force_dbt_deps=force_dbt_deps - ) - - -def test_data_monitoring_internal_runner_vars(config, mock_create_dbt_runner): - data_monitoring = DataMonitoring.__new__(DataMonitoring) - data_monitoring.config = config - data_monitoring.force_update_dbt_package = True - data_monitoring._init_internal_dbt_runner() - _assert_internal_runner_created(mock_create_dbt_runner, config, force_dbt_deps=True) - - -def test_debug_internal_runner_vars(config, mock_create_dbt_runner): - assert Debug(config).run() - _assert_internal_runner_created(mock_create_dbt_runner, config) - - -def test_upload_source_freshness_internal_runner_vars(config, mock_create_dbt_runner): - UploadSourceFreshnessOperation(config).upload_results( - results=[], metadata={"invocation_id": "id"}, rows_per_insert=10 - ) - _assert_internal_runner_created(mock_create_dbt_runner, config) - - -@mock.patch("subprocess.run") -def test_runner_vars_merged_with_per_call_vars(mock_subprocess_run): - runner = SubprocessDbtRunner( - project_dir="proj_dir", - profiles_dir="prof_dir", - vars={"query_max_size": 1000, "days_back": 30}, - run_deps_if_needed=False, - ) - runner.run(select="model", vars={"days_back": 7}) - args = mock_subprocess_run.call_args[0][0] - assert json.loads(args[args.index("--vars") + 1]) == { - "query_max_size": 1000, - "days_back": 7, - } - - -def test_populate_alerts_uses_runner_vars(): - data_monitoring_alerts = DataMonitoringAlertsMock() - with mock.patch.object( - data_monitoring_alerts.internal_dbt_runner, "run", return_value=True - ) as run: - assert data_monitoring_alerts._populate_data() - run.assert_called_once_with( - select="elementary_cli.alerts.alerts_v2", - full_refresh=False, - ) From caf4a3be7a6215fd71b1943a7fbad0c922c81f08 Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 13:29:57 +0000 Subject: [PATCH 7/9] Simplify days_back injection in Config --- elementary/config/config.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/elementary/config/config.py b/elementary/config/config.py index 0d7c45083..565644383 100644 --- a/elementary/config/config.py +++ b/elementary/config/config.py @@ -92,9 +92,10 @@ def __init__( self.env_vars = dict() self.env_vars.update(self._parse_dbt_quoting_to_env_vars(dbt_quoting)) - self.dbt_vars = ( - {**(dbt_vars or {}), "days_back": days_back} if days_back else dbt_vars - ) + self.dbt_vars = dbt_vars + if days_back is not None: + self.dbt_vars = self.dbt_vars or {} + self.dbt_vars["days_back"] = days_back config = self._load_configuration() From 12f61fe6ee3f64a67e9a0f207152af3f98e14302 Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 13:32:40 +0000 Subject: [PATCH 8/9] Keep alerts_v2 default days_back when --days-back is 0 --- elementary/config/config.py | 3 ++- tests/unit/cli/test_dbt_vars_option.py | 19 ++++++++++++------- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/elementary/config/config.py b/elementary/config/config.py index 565644383..5ad55d881 100644 --- a/elementary/config/config.py +++ b/elementary/config/config.py @@ -93,7 +93,8 @@ def __init__( self.env_vars.update(self._parse_dbt_quoting_to_env_vars(dbt_quoting)) self.dbt_vars = dbt_vars - if days_back is not None: + # alerts_v2 populates nothing with days_back=0, so 0 keeps its default of 1. + if days_back: self.dbt_vars = self.dbt_vars or {} self.dbt_vars["days_back"] = days_back diff --git a/tests/unit/cli/test_dbt_vars_option.py b/tests/unit/cli/test_dbt_vars_option.py index 21112d10d..c2650e24b 100644 --- a/tests/unit/cli/test_dbt_vars_option.py +++ b/tests/unit/cli/test_dbt_vars_option.py @@ -34,15 +34,20 @@ def _invoke(cli, args: List[str]): @pytest.mark.parametrize( - "extra_args,expected_dbt_vars", + "extra_args,expected_dbt_vars,expected_days_back", [ - ([], {"days_back": 1}), - (["--dbt-vars", DBT_VARS], {**EXPECTED_DBT_VARS, "days_back": 1}), - (["--days-back", "3"], {"days_back": 3}), - (["--dbt-vars", "{days_back: 10}", "--days-back", "3"], {"days_back": 3}), + ([], {"days_back": 1}, 1), + (["--dbt-vars", DBT_VARS], {**EXPECTED_DBT_VARS, "days_back": 1}, 1), + (["--days-back", "3"], {"days_back": 3}, 3), + (["--days-back", "0"], None, 0), + ( + ["--dbt-vars", "{days_back: 10}", "--days-back", "3"], + {"days_back": 3}, + 3, + ), ], ) -def test_monitor_dbt_vars(tmp_path, extra_args, expected_dbt_vars): +def test_monitor_dbt_vars(tmp_path, extra_args, expected_dbt_vars, expected_days_back): with mock.patch( "elementary.monitor.cli.DataMonitoringAlerts" ) as data_monitoring_alerts: @@ -53,7 +58,7 @@ def test_monitor_dbt_vars(tmp_path, extra_args, expected_dbt_vars): config = data_monitoring_alerts.call_args.kwargs["config"] assert config.dbt_vars == expected_dbt_vars data_monitoring_alerts.return_value.run_alerts.assert_called_once_with( - expected_dbt_vars["days_back"], False + expected_days_back, False ) From b629331bf99f4d236aa2f2d75068ed822993156e Mon Sep 17 00:00:00 2001 From: guy Date: Tue, 6 Oct 2026 13:38:50 +0000 Subject: [PATCH 9/9] Drop days_back comment in Config --- elementary/config/config.py | 1 - 1 file changed, 1 deletion(-) diff --git a/elementary/config/config.py b/elementary/config/config.py index 5ad55d881..3d30ca0dc 100644 --- a/elementary/config/config.py +++ b/elementary/config/config.py @@ -93,7 +93,6 @@ def __init__( self.env_vars.update(self._parse_dbt_quoting_to_env_vars(dbt_quoting)) self.dbt_vars = dbt_vars - # alerts_v2 populates nothing with days_back=0, so 0 keeps its default of 1. if days_back: self.dbt_vars = self.dbt_vars or {} self.dbt_vars["days_back"] = days_back