fix(providers/common/sql): add dummy connection setter for backward c…

…ompatibility (apache#42490)

the introduction of connection property breaks apache-airflow-providers-mysql<5.7.1,
apache-airflow-providers-elasticsearch<5.5.1
 and apache-airflow-providers-postgres<5.13.0

airflow/providers/common/sql/hooks/sql.py

@@ -210,6 +210,17 @@ def connection(self) -> Connection:
             self._connection = self.get_connection(self.get_conn_id())
         return self._connection
 
+    @connection.setter
+    def connection(self, value: Any) -> None:
+        # This setter is for backward compatibility and should not be used.
+        # Since the introduction of connection property, the providers listed below
+        # breaks due to assigning value to self.connection
+        #
+        # apache-airflow-providers-mysql<5.7.1
+        # apache-airflow-providers-elasticsearch<5.5.1
+        # apache-airflow-providers-postgres<5.13.0
+        pass
+
     @property
     def connection_extra(self) -> dict:
         return self.connection.extra_dejson

airflow/providers/common/sql/hooks/sql.pyi

@@ -70,6 +70,8 @@ class DbApiHook(BaseHook):
     def placeholder(self): ...
     @property
     def connection(self) -> Connection: ...
+    @connection.setter
+    def connection(self, value: Any) -> None: ...
     @property
     def connection_extra(self) -> dict: ...
     @cached_property

tests/providers/mysql/hooks/test_mysql.py

@@ -67,6 +67,21 @@ def test_get_conn(self, mock_connect):
         assert kwargs["host"] == "host"
         assert kwargs["db"] == "schema"
 
+    @mock.patch("MySQLdb.connect")
+    def test_dummy_connection_setter(self, mock_connect):
+        self.db_hook.get_conn()
+
+        self.db_hook.connection = "Won't affect anything"
+        assert self.db_hook.connection != "Won't affect anything"
+
+        assert mock_connect.call_count == 1
+        args, kwargs = mock_connect.call_args
+        assert args == ()
+        assert kwargs["user"] == "login"
+        assert kwargs["passwd"] == "password"
+        assert kwargs["host"] == "host"
+        assert kwargs["db"] == "schema"
+
     @mock.patch("MySQLdb.connect")
     def test_get_uri(self, mock_connect):
         self.connection.extra = json.dumps({"charset": "utf-8"})