Provider Migration: Update mysql for Airflow 3.0 compatibility (#52500)

Follow-up of #52292. Part of #52378

Author: kgw7401

providers/mysql/src/airflow/providers/mysql/transfers/presto_to_mysql.py

@@ -20,8 +20,8 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
-from airflow.models import BaseOperator
 from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.providers.mysql.version_compat import BaseOperator
 from airflow.providers.presto.hooks.presto import PrestoHook
 
 if TYPE_CHECKING:

providers/mysql/src/airflow/providers/mysql/transfers/s3_to_mysql.py

@@ -20,9 +20,9 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
-from airflow.models import BaseOperator
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.providers.mysql.version_compat import BaseOperator
 
 if TYPE_CHECKING:
     try:

providers/mysql/src/airflow/providers/mysql/transfers/trino_to_mysql.py

@@ -20,8 +20,8 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
-from airflow.models import BaseOperator
 from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.providers.mysql.version_compat import BaseOperator
 from airflow.providers.trino.hooks.trino import TrinoHook
 
 if TYPE_CHECKING:

providers/mysql/src/airflow/providers/mysql/transfers/vertica_to_mysql.py

@@ -33,8 +33,8 @@
         "installed in case you see compilation error during installation."
     )
 
-from airflow.models import BaseOperator
 from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.providers.mysql.version_compat import BaseOperator
 from airflow.providers.vertica.hooks.vertica import VerticaHook
 
 if TYPE_CHECKING:

providers/mysql/src/airflow/providers/mysql/version_compat.py

@@ -0,0 +1,40 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+
+def get_base_airflow_version_tuple() -> tuple[int, int, int]:
+    from packaging.version import Version
+
+    from airflow import __version__
+
+    airflow_version = Version(__version__)
+    return airflow_version.major, airflow_version.minor, airflow_version.micro
+
+
+AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
+
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import BaseOperator
+else:
+    from airflow.models import BaseOperator
+
+__all__ = [
+    "AIRFLOW_V_3_0_PLUS",
+    "BaseOperator",
+]

providers/mysql/tests/unit/mysql/transfers/test_trino_to_mysql.py

@@ -65,16 +65,21 @@ def test_execute_with_mysql_preoperator(self, mock_trino_hook, mock_mysql_hook):
     @pytest.mark.skipif(
         "AIRFLOW_RUNALL_TESTS" not in os.environ, reason="Skipped because AIRFLOW_RUNALL_TESTS is not set"
     )
-    def test_trino_to_mysql(self):
-        op = TrinoToMySqlOperator(
-            task_id="trino_to_mysql_check",
-            sql="""
-                SELECT name, count(*) as ccount
-                FROM airflow.static_babynames
-                GROUP BY name
-                """,
-            mysql_table="test_static_babynames",
-            mysql_preoperator="TRUNCATE TABLE test_static_babynames;",
-            dag=self.dag,
-        )
-        op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
+    def test_trino_to_mysql(self, dag_maker):
+        with dag_maker(
+            dag_id="test_trino_to_mysql_transfer",
+            schedule=None,
+            start_date=DEFAULT_DATE,
+        ):
+            TrinoToMySqlOperator(
+                task_id="trino_to_mysql_check",
+                sql="""
+                    SELECT name, count(*) as ccount
+                    FROM airflow.static_babynames
+                    GROUP BY name
+                    """,
+                mysql_table="test_static_babynames",
+                mysql_preoperator="TRUNCATE TABLE test_static_babynames;",
+            )
+        dr = dag_maker.create_dagrun()
+        dag_maker.run_ti("trino_to_mysql_check", dr)