Enhance Variable set method to use upsert instead of delsert

airflow-core/src/airflow/models/variable.py

@@ -186,7 +186,7 @@ def set(
         """
         Set a value for an Airflow Variable with a given Key.
 
-        This operation overwrites an existing variable.
+        This operation overwrites an existing variable using the session's dialect-specific upsert operation.
 
         :param key: Variable Key
         :param value: Value to set for the Variable
@@ -231,9 +231,47 @@ def set(
             ctx = create_session()
 
         with ctx as session:
-            Variable.delete(key, session=session)
-            session.add(Variable(key=key, val=stored_value, description=description))
-            session.flush()
+            new_variable = Variable(key=key, val=stored_value, description=description)
+
+            val = new_variable._val
+            is_encrypted = new_variable.is_encrypted
+
+            # Import dialect-specific insert function
+            if (dialect_name := session.get_bind().dialect.name) == "postgresql":
+                from sqlalchemy.dialects.postgresql import insert
+            elif dialect_name == "mysql":
+                from sqlalchemy.dialects.mysql import insert
+            else:
+                from sqlalchemy.dialects.sqlite import insert
+
+            # Create the insert statement (common for all dialects)
+            stmt = insert(Variable).values(
+                key=key,
+                val=val,
+                description=description,
+                is_encrypted=is_encrypted,
+            )
+
+            # Apply dialect-specific upsert
+            if dialect_name == "mysql":
+                # MySQL: ON DUPLICATE KEY UPDATE
+                stmt = stmt.on_duplicate_key_update(
+                    val=val,
+                    description=description,
+                    is_encrypted=is_encrypted,
+                )
+            else:
+                # PostgreSQL and SQLite: ON CONFLICT DO UPDATE
+                stmt = stmt.on_conflict_do_update(
+                    index_elements=["key"],
+                    set_=dict(
+                        val=val,
+                        description=description,
+                        is_encrypted=is_encrypted,
+                    ),
+                )
+
+            session.execute(stmt)
             # invalidate key in cache for faster propagation
             # we cannot save the value set because it's possible that it's shadowed by a custom backend
             # (see call to check_for_write_conflict above)

airflow-core/tests/unit/models/test_variable.py

@@ -121,6 +121,27 @@ def test_variable_set_with_env_variable(self, caplog, session):
             "EnvironmentVariablesBackend"
         )
 
+    def test_variable_set_update_existing(self, session):
+        Variable.set(key="test_key", value="initial_value", session=session)
+
+        initial_var = session.query(Variable).filter(Variable.key == "test_key").one()
+        initial_id = initial_var.id
+
+        # Need to expire session cache to fetch fresh data from db on next query
+        # Without this, SQLAlchemy will return the cached object with old values
+        # instead of querying the database again for the updated values
+        session.expire(initial_var)
+
+        Variable.set(key="test_key", value="updated_value", session=session)
+
+        updated_var = session.query(Variable).filter(Variable.key == "test_key").one()
+
+        # 1. The ID remains the same (no delete-insert)
+        assert updated_var.id == initial_id, "Variable ID should remain the same after update"
+
+        # 2. The value is updated to the new value
+        assert updated_var.val == "updated_value", "Variable value should be updated to the new value"
+
     @mock.patch("airflow.models.variable.ensure_secrets_loaded")
     def test_variable_set_with_extra_secret_backend(self, mock_ensure_secrets, caplog, session):
         caplog.set_level(logging.WARNING, logger=variable.log.name)