Fix MWAA client's set_variable (#1524)

* Fix MWAA client's set_variable

We need to escape quotes and escape sequences within the value of this function so Airflow can recognize the string

Additionally, when we try to quote (and escape) with single quotes, Airflow returns a 500 error. We guess this is because Airflow is using escaped single quotes as part of its internal functioning

Author: max-winderbaum

sqlmesh/schedulers/airflow/mwaa_client.py

@@ -31,7 +31,8 @@ def set_variable(self, key: str, value: str) -> t.Tuple[str, str]:
         Returns:
             A tuple of stdout and stderr from the MWAA CLI.
         """
-        return self._post(f"variables set {key} '{value}'")
+        value = value.replace("\\", "\\\\").replace('"', '\\"')
+        return self._post(f'variables set {key} "{value}"')
 
     def get_first_dag_run_id(self, dag_id: str) -> t.Optional[str]:
         dag_runs = self._list_dag_runs(dag_id)

tests/schedulers/airflow/test_mwaa_client.py

@@ -22,7 +22,7 @@ def test_set_variable(mocker: MockerFixture):
 
     set_variable_mock.assert_called_once_with(
         "https://test_airflow_host/aws_mwaa/cli",
-        data="variables set test_key 'test_value'",
+        data='variables set test_key "test_value"',
     )