Warn about precedence of env var when geting variables

Author: Kamil Breguła

airflow/models/variable.py

@@ -17,6 +17,8 @@
 # under the License.
 
 import json
+import logging
+import os
 from typing import Any, Optional
 
 from cryptography.fernet import InvalidToken as InvalidFernetToken
@@ -30,6 +32,8 @@
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.session import provide_session
 
+log = logging.getLogger()
+
 
 class Variable(Base, LoggingMixin):
     """
@@ -143,6 +147,14 @@ def set(cls, key: str, value: Any, serialize_json: bool = False, session: Sessio
         :param serialize_json: Serialize the value to a JSON string
         :param session: SQL Alchemy Sessions
         """
+        env_var_name = "AIRFLOW_VAR_" + key.upper()
+        if env_var_name in os.environ:
+            log.warning(
+                "You have the environment variable %s defined, which takes precedence over reading "
+                "from the database. The value will be saved, but to read it you have to delete "
+                "the environment variable.",
+                env_var_name,
+            )
         if serialize_json:
             stored_value = json.dumps(value, indent=2)
         else:

tests/models/test_variable.py

@@ -17,12 +17,13 @@
 # under the License.
 
 import unittest
+from unittest import mock
 
 from cryptography.fernet import Fernet
 from parameterized import parameterized
 
 from airflow import settings
-from airflow.models import Variable, crypto
+from airflow.models import Variable, crypto, variable
 from tests.test_utils import db
 from tests.test_utils.config import conf_vars
 
@@ -89,6 +90,23 @@ def test_variable_set_get_round_trip(self):
         Variable.set("tested_var_set_id", "Monday morning breakfast")
         self.assertEqual("Monday morning breakfast", Variable.get("tested_var_set_id"))
 
+    def test_variable_set_with_env_variable(self):
+        Variable.set("key", "db-value")
+        with self.assertLogs(variable.log) as log_context:
+            with mock.patch.dict('os.environ', AIRFLOW_VAR_KEY="env-value"):
+                Variable.set("key", "new-db-value")
+                self.assertEqual("env-value", Variable.get("key"))
+            self.assertEqual("new-db-value", Variable.get("key"))
+
+        self.assertEqual(
+            log_context.records[0].message,
+            (
+                'You have the environment variable AIRFLOW_VAR_KEY defined, which takes precedence over '
+                'reading from the database. The value will be saved, but to read it you have to delete '
+                'the environment variable.'
+            ),
+        )
+
     def test_variable_set_get_round_trip_json(self):
         value = {"a": 17, "b": 47}
         Variable.set("tested_var_set_id", value, serialize_json=True)