Repository: TobikoData/sqlmesh
Pull request #3592 (2025-01-07 16:17:37)

Fix: change SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION var name
fixes https://github.com/TobikoData/sqlmesh/issues/3573

since `load_config_from_env` method assumes that the sqlmesh config vars starts with `"SQLMESH__"`, the `SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION` var name is misinterpreted as an input for config and it throws below error,

```
pydantic_core._pydantic_core.ValidationError: 1 validation error for Config
Extra inputs are not permitted [type=extra_forbidden, input_value={'disable_state_migration': '1'}, input_type=dict]
```
hence the variable name has been renamed.

I believe there are more than one solution to solve this issue, but I feel this is more sensible and a simple change. In fact, I've created a custom class for the SQLMesh generic context with an arg to not load env vars as config as a workaround for my current project. However, I really appreciate any feedbacks or suggestions and happy to contribute here. Thanks :)

izeigerman: @lafirmm this would be a breaking change. Can you describe how you run into this issue? I don't think that SQLMesh configuration is expected to be loaded on the Airflow side
lafirmm: > @lafirmm this is a breaking change. Can you describe how you ran into this issue? I don't think that SQLMesh configuration is expected to be loaded on the Airflow side

I want to handle the SQLMesh state migration by myself and not done by this [SqlmeshAirflowPlugin](https://github.com/TobikoData/sqlmesh/blob/9d7c3856fd77e3a1be86f04e0b994c6186e70f39/sqlmesh/schedulers/airflow/plugin.py#L59) initaited by this [setup](https://github.com/TobikoData/sqlmesh/blob/9d7c3856fd77e3a1be86f04e0b994c6186e70f39/setup.py#L25).
Hence I created an environment variable in airflow containers (managed by our team) called `SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION` with value `1` as described and introduced here https://github.com/TobikoData/sqlmesh/pull/2921, and the environment variable I created `SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION` in airflow containers is misinterpreted as an input for config and it throws above error.

izeigerman: > And, the SQLMesh config is loaded on the airflow side because we don't use built-in dag generators (SQLMeshAirflow class) but a custom method.

I think this is a problem then.

> Also can you please explain how this is a breaking change since this env var name SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION is used nowhere except [here](https://github.com/TobikoData/sqlmesh/blob/9d7c3856fd77e3a1be86f04e0b994c6186e70f39/sqlmesh/schedulers/airflow/plugin.py#L34).

Because existing users are relying on it
izeigerman: To ensure that we don't break existing users I'd suggest excluding this variable from the config loader, even though it's a hackier approach
lafirmm: > To ensure that we don't break existing users I'd suggest excluding this variable from the config loader, even though it's a hackier approach

@izeigerman I agree and made the corresponding changes, I'd really appreciate your feedback

Patch:
diff --git a/sqlmesh/core/config/loader.py b/sqlmesh/core/config/loader.py
index a75eacb0..2faaed5e 100644
--- a/sqlmesh/core/config/loader.py
+++ b/sqlmesh/core/config/loader.py
@@ -175,7 +175,7 @@ def load_config_from_env() -> t.Dict[str, t.Any]:
 
     for key, value in os.environ.items():
         key = key.lower()
-        if key.startswith(f"{c.SQLMESH}__"):
+        if key.startswith(f"{c.SQLMESH}__") and key != (c.DISABLE_SQLMESH_STATE_MIGRATION).lower():
             segments = key.split("__")[1:]
             if not segments or not segments[-1]:
                 raise ConfigError(f"Invalid SQLMesh configuration variable '{key}'.")
diff --git a/sqlmesh/core/constants.py b/sqlmesh/core/constants.py
index 3fcb135f..df204cfb 100644
--- a/sqlmesh/core/constants.py
+++ b/sqlmesh/core/constants.py
@@ -88,3 +88,5 @@ AIRFLOW = "airflow"
 DBT = "dbt"
 NATIVE = "native"
 HYBRID = "hybrid"
+
+DISABLE_SQLMESH_STATE_MIGRATION = "SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION"
diff --git a/sqlmesh/schedulers/airflow/plugin.py b/sqlmesh/schedulers/airflow/plugin.py
index 9ecb0fbc..6813bcc3 100644
--- a/sqlmesh/schedulers/airflow/plugin.py
+++ b/sqlmesh/schedulers/airflow/plugin.py
@@ -31,7 +31,7 @@ class SqlmeshAirflowPlugin(AirflowPlugin):
             logger.info("MWAA Webserver instance detected. Skipping SQLMesh state migration...")
             return
 
-        if str_to_bool(os.environ.get("SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION", "0")):
+        if str_to_bool(os.environ.get(c.DISABLE_SQLMESH_STATE_MIGRATION, "0")):
             logger.info("SQLMesh state migration disabled. Must be handled outside of Airflow")
             return
 


Test patch:
diff --git a/tests/core/test_config.py b/tests/core/test_config.py
index 42b23472..b8696902 100644
--- a/tests/core/test_config.py
+++ b/tests/core/test_config.py
@@ -302,6 +302,17 @@ def test_load_config_from_env():
         )
 
 
+def test_load_config_from_env_no_config_vars():
+    with mock.patch.dict(
+        os.environ,
+        {
+            "SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION": "1",
+            "DUMMY_ENV_VAR": "dummy",
+        },
+    ):
+        assert load_config_from_env() == {}
+
+
 def test_load_config_from_env_invalid_variable_name():
     with mock.patch.dict(
         os.environ,