Switch the default async MySQL driver to asyncmy (#75218)

Author: Dev-iL

airflow-core/docs/howto/set-up-database.rst

@@ -373,7 +373,7 @@ We recommend using the ``mysqlclient`` driver and specifying it in your SqlAlche
     mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
 
 .. important::
-    The integration of MySQL backend has only been validated using the ``mysqlclient`` driver
+    The synchronous integration of MySQL backend has only been validated using the ``mysqlclient`` driver
     during Apache Airflow's continuous integration (CI) process.
 
 If you want to use other drivers visit the `MySQL Dialect <https://docs.sqlalchemy.org/en/20/dialects/mysql.html>`__  in SQLAlchemy documentation for more information regarding download
@@ -389,6 +389,32 @@ In addition, you also should pay particular attention to MySQL's encoding. Altho
     Read https://stackoverflow.com/questions/9192027/invalid-default-value-for-create-date-timestamp-field for how to disable it.
     See `SQL Mode - NO_ZERO_DATE <https://dev.mysql.com/doc/refman/8.0/en/sql-mode.html#sqlmode_no_zero_date>`__ for more information.
 
+Async SQLAlchemy engine and the async driver
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+When ``[database] sql_alchemy_conn_async`` is not set, Airflow derives the async MySQL
+metadata database URL from ``sql_alchemy_conn``. It selects ``asyncmy`` when installed:
+
+.. code-block:: text
+
+   mysql+asyncmy://<user>:<password>@<host>[:<port>]/<dbname>
+
+The MySQL provider installs ``asyncmy`` by default. If an older provider release does not
+install it, core falls back to ``aiomysql``, so upgrading core alone keeps working.
+The provider also keeps installing ``aiomysql`` for older supported core releases that
+always derive a ``mysql+aiomysql://`` URL.
+
+An explicit ``sql_alchemy_conn_async`` is never rewritten. To keep using ``aiomysql``,
+install ``apache-airflow-providers-mysql[aiomysql]`` and set:
+
+.. code-block:: ini
+
+   [database]
+   sql_alchemy_conn_async = mysql+aiomysql://<user>:<password>@<host>[:<port>]/<dbname>
+
+Review ``sql_alchemy_connect_args_async`` when changing drivers: driver-specific
+connection arguments must be supported by the selected driver.
+
 MsSQL Database
 --------------
 

airflow-core/newsfragments/75218.significant.rst

@@ -0,0 +1,13 @@
+Default async MySQL driver changed from aiomysql to asyncmy
+
+When ``[database] sql_alchemy_conn_async`` is not set, Airflow now derives a
+``mysql+asyncmy://`` metadata database URL when asyncmy is installed. The MySQL
+provider installs asyncmy by default. If asyncmy is absent, core keeps deriving
+a ``mysql+aiomysql://`` URL, preserving compatibility with older provider releases.
+
+The provider retains aiomysql as a default dependency for older supported core
+releases that cannot select asyncmy. An explicit ``sql_alchemy_conn_async`` is
+never rewritten. To keep using aiomysql, install
+``apache-airflow-providers-mysql[aiomysql]`` and configure a
+``mysql+aiomysql://`` async URL. Review driver-specific
+``sql_alchemy_connect_args_async`` values when switching drivers.

airflow-core/src/airflow/settings.py

@@ -26,6 +26,7 @@
 from collections.abc import Callable, Mapping
 from functools import cache, partial
 from importlib import metadata
+from importlib.util import find_spec
 from typing import TYPE_CHECKING, Any, Literal
 
 import pluggy
@@ -63,13 +64,13 @@
 
 _USE_PSYCOPG3: bool
 try:
-    from importlib.util import find_spec
-
     _USE_PSYCOPG3 = find_spec("psycopg") is not None
 
 except (ImportError, ModuleNotFoundError):
     _USE_PSYCOPG3 = False
 
+_USE_ASYNCMY = find_spec("asyncmy") is not None
+
 if TYPE_CHECKING:
     from sqlalchemy.engine import Engine
 
@@ -266,7 +267,7 @@ def _get_async_conn_uri_from_sync(sync_uri):
     AIO_LIBS_MAPPING = {
         "sqlite": "aiosqlite",
         "postgresql": "psycopg_async" if _USE_PSYCOPG3 else "asyncpg",
-        "mysql": "aiomysql",
+        "mysql": "asyncmy" if _USE_ASYNCMY else "aiomysql",
     }
 
     scheme, rest = sync_uri.split(":", maxsplit=1)

airflow-core/tests/unit/core/test_settings.py

@@ -545,7 +545,6 @@ def test_sqlite_relative_path(value, expectation):
         ("postgresql+psycopg2://user:pass@host/db", True, "postgresql+psycopg_async://user:pass@host/db"),
         ("postgresql+psycopg2://user:pass@host/db", False, "postgresql+asyncpg://user:pass@host/db"),
         ("sqlite:////root/airflow.db", True, "sqlite+aiosqlite:////root/airflow.db"),
-        ("mysql://user:pass@host/db", True, "mysql+aiomysql://user:pass@host/db"),
         ("mssql://user:pass@host/db", True, "mssql://user:pass@host/db"),
     ],
 )
@@ -556,10 +555,28 @@ def test_get_async_conn_uri_from_sync(sync_uri, use_psycopg3, expected_async_uri
         assert settings._get_async_conn_uri_from_sync(sync_uri) == expected_async_uri
 
 
-def test_explicit_sql_alchemy_conn_async_is_not_rewritten():
+@pytest.mark.parametrize("sync_scheme", ["mysql", "mysql+mysqldb", "mysql+pymysql"])
+@pytest.mark.parametrize(("use_asyncmy", "expected_driver"), [(True, "asyncmy"), (False, "aiomysql")])
+def test_get_async_mysql_conn_uri_from_sync(monkeypatch, sync_scheme, use_asyncmy, expected_driver):
+    monkeypatch.setattr(settings, "_USE_ASYNCMY", use_asyncmy)
+    rest = "user:p%40ss@host:3307/db?charset=utf8mb4&unix_socket=%2Ftmp%2Fmysql.sock"
+
+    assert settings._get_async_conn_uri_from_sync(f"{sync_scheme}://{rest}") == (
+        f"mysql+{expected_driver}://{rest}"
+    )
+
+
+@pytest.mark.parametrize(
+    "explicit_async_uri",
+    [
+        "postgresql+asyncpg://user:pass@host/db",
+        "mysql+aiomysql://user:pass@host/db",
+        "mysql+asyncmy://user:pass@host/db",
+    ],
+)
+def test_explicit_sql_alchemy_conn_async_is_not_rewritten(explicit_async_uri):
     from airflow import settings
 
-    explicit_async_uri = "postgresql+asyncpg://user:pass@host/db"
     try:
         with conf_vars({("database", "sql_alchemy_conn_async"): explicit_async_uri}):
             settings.configure_vars()
@@ -693,6 +710,7 @@ def test_failed_disposal_preserves_resources_for_retry(self, monkeypatch, error)
         pytest.param("postgresql+psycopg_async", marks=pytest.mark.backend("postgres")),
         pytest.param("postgresql+asyncpg", marks=pytest.mark.backend("postgres")),
         pytest.param("mysql+aiomysql", marks=pytest.mark.backend("mysql")),
+        pytest.param("mysql+asyncmy", marks=pytest.mark.backend("mysql")),
         pytest.param("sqlite+aiosqlite", marks=pytest.mark.backend("sqlite")),
     ],
 )
@@ -723,6 +741,8 @@ async def run():
                     pass
                 else:
                     raise AssertionError("connection remained open")
+            elif driver == "mysql+asyncmy":
+                assert not raw.connected
             else:
                 assert raw.is_closed() if driver == "postgresql+asyncpg" else raw.closed
 

providers/mysql/README.rst

@@ -59,6 +59,7 @@ PIP package                                 Version required
 ``mysqlclient``                             ``>=2.2.5; sys_platform != "darwin"``
 ``mysql-connector-python``                  ``>=9.1.0; python_version < "3.12"``
 ``mysql-connector-python``                  ``>=9.1.0,!=9.7.0; python_version >= "3.12"``
+``asyncmy``                                 ``>=0.2.16``
 ``aiomysql``                                ``>=0.2.0``
 ``pymysql``                                 ``>=1.0.3,<1.2``
 ==========================================  =============================================
@@ -92,6 +93,7 @@ Optional dependencies
 ==========================  ========================================
 Extra                       Dependencies
 ==========================  ========================================
+``aiomysql``                ``aiomysql>=0.2.0``
 ``mysql-connector-python``
 ``amazon``                  ``apache-airflow-providers-amazon``
 ``openlineage``             ``apache-airflow-providers-openlineage``

providers/mysql/docs/changelog.rst

@@ -26,6 +26,17 @@
 Changelog
 ---------
 
+.. note::
+
+   The provider now installs ``asyncmy`` for the async metadata database engine.
+   Core releases that include asyncmy selection derive ``mysql+asyncmy://`` URLs
+   when the driver is installed. Older core releases continue to use ``aiomysql``,
+   which remains installed by default for compatibility.
+
+   To explicitly select the previous driver, set ``[database] sql_alchemy_conn_async``
+   to a ``mysql+aiomysql://`` URL. The ``[aiomysql]`` extra also declares this driver
+   for installations that need it after it becomes optional.
+
 6.7.1
 .....
 

providers/mysql/docs/index.rst

@@ -106,6 +106,7 @@ PIP package                                 Version required
 ``mysqlclient``                             ``>=2.2.5; sys_platform != "darwin"``
 ``mysql-connector-python``                  ``>=9.1.0; python_version < "3.12"``
 ``mysql-connector-python``                  ``>=9.1.0,!=9.7.0; python_version >= "3.12"``
+``asyncmy``                                 ``>=0.2.16``
 ``aiomysql``                                ``>=0.2.0``
 ``pymysql``                                 ``>=1.0.3,<1.2``
 ==========================================  =============================================
@@ -141,12 +142,13 @@ Install them when installing from PyPI. For example:
 
 .. code-block:: bash
 
-    pip install apache-airflow-providers-mysql[mysql-connector-python]
+    pip install apache-airflow-providers-mysql[aiomysql]
 
 
 ==========================  ========================================
 Extra                       Dependencies
 ==========================  ========================================
+``aiomysql``                ``aiomysql>=0.2.0``
 ``mysql-connector-python``
 ``amazon``                  ``apache-airflow-providers-amazon``
 ``openlineage``             ``apache-airflow-providers-openlineage``

providers/mysql/pyproject.toml

@@ -67,17 +67,23 @@ dependencies = [
     'mysqlclient>=2.2.5; sys_platform != "darwin"',
     'mysql-connector-python>=9.1.0; python_version < "3.12"',
     'mysql-connector-python>=9.1.0, !=9.7.0; python_version >= "3.12"',
+    "asyncmy>=0.2.16",
+    # Older supported Airflow cores derive mysql+aiomysql URLs unconditionally.
+    # Keep aiomysql installed until every supported core can select asyncmy.
+    "aiomysql>=0.2.0",
     # pymysql 1.2.0 changed ping() to require reconnect as a positional arg;
     # SQLAlchemy's AsyncAdapt_aiomysql_connection.ping has no default for it — cap
     # until the SQLAlchemy fix is released. Tracked upstream at
     # https://github.com/sqlalchemy/sqlalchemy/issues/13306
-    "aiomysql>=0.2.0",
     "pymysql>=1.0.3,<1.2",
 ]
 
 # The optional dependencies should be modified in place in the generated file
 # Any change in the dependencies is preserved when the file is regenerated
 [project.optional-dependencies]
+"aiomysql" = [
+    "aiomysql>=0.2.0",
+]
 # only needed for backwards compatibility
 "mysql-connector-python" = [
 ]