From 22ea45b875753a183acc2e53207d2377f86c0692 Mon Sep 17 00:00:00 2001 From: Gayathri Srividya Rajavarapu Date: Sat, 20 Jun 2026 16:33:02 +0530 Subject: [PATCH 1/3] Show clear error for invalid sql_alchemy_conn at startup Users can hit a cryptic unpacking ValueError when sql_alchemy_conn is malformed, especially while diagnosing slow CLI startup in constrained environments. A clearer validation path makes misconfiguration immediately actionable and reduces support/debugging churn. --- airflow-core/src/airflow/settings.py | 23 +++++++- airflow-core/tests/unit/core/test_settings.py | 55 +++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/settings.py b/airflow-core/src/airflow/settings.py index 54150dd8bfd5b..148254f86c51c 100644 --- a/airflow-core/src/airflow/settings.py +++ b/airflow-core/src/airflow/settings.py @@ -241,13 +241,34 @@ def load_policy_plugins(pm: pluggy.PluginManager): def _get_async_conn_uri_from_sync(sync_uri): - # Mapping of backend to async driver: + """ + Convert a sync SQLAlchemy URI to its async equivalent. + + Args: + sync_uri: The synchronous SQLAlchemy connection URI + + Returns: + The async URI with appropriate async driver (e.g., aiosqlite, asyncpg, aiomysql), + or the original URI if no async driver mapping exists + + Raises: + ValueError: If the URI is malformed (missing ':' scheme separator) + """ + # Mapping of backend to async driver. AIO_LIBS_MAPPING = { "sqlite": "aiosqlite", "postgresql": "psycopg_async" if _USE_PSYCOPG3 else "asyncpg", "mysql": "aiomysql", } + if not sync_uri or ":" not in sync_uri: + raise ValueError( + f"Invalid SQLAlchemy connection URI: {sync_uri!r}. " + "The URI must be in the format 'scheme://host/path' or similar with a ':' separator. " + "Check that AIRFLOW__DATABASE__SQL_ALCHEMY_CONN environment variable or " + "sql_alchemy_conn in airflow.cfg contains a valid database URI." + ) + scheme, rest = sync_uri.split(":", maxsplit=1) scheme = scheme.split("+", maxsplit=1)[0] aiolib = AIO_LIBS_MAPPING.get(scheme) diff --git a/airflow-core/tests/unit/core/test_settings.py b/airflow-core/tests/unit/core/test_settings.py index c703ee05ab941..126610f1ce944 100644 --- a/airflow-core/tests/unit/core/test_settings.py +++ b/airflow-core/tests/unit/core/test_settings.py @@ -533,3 +533,58 @@ def test_early_return_when_all_none(self): settings.dispose_orm(do_log=False) mock_close.assert_not_called() + + +class TestGetAsyncConnUriFromSync: + """Tests for _get_async_conn_uri_from_sync function.""" + + def test_sqlite_uri_conversion(self): + """Test conversion of SQLite sync URI to async with aiosqlite.""" + result = settings._get_async_conn_uri_from_sync("sqlite:///path/to/db.sqlite") + assert result == "sqlite+aiosqlite:///path/to/db.sqlite" + + def test_postgresql_uri_conversion(self): + """Test conversion of PostgreSQL sync URI to async with asyncpg.""" + result = settings._get_async_conn_uri_from_sync("postgresql://user:pass@localhost/dbname") + assert result == "postgresql+asyncpg://user:pass@localhost/dbname" + + def test_mysql_uri_conversion(self): + """Test conversion of MySQL sync URI to async with aiomysql.""" + result = settings._get_async_conn_uri_from_sync("mysql://user:pass@localhost/dbname") + assert result == "mysql+aiomysql://user:pass@localhost/dbname" + + def test_postgresql_psycopg2_uri_conversion(self): + """Test conversion of PostgreSQL with psycopg2 driver to asyncpg.""" + result = settings._get_async_conn_uri_from_sync("postgresql+psycopg2://user@localhost/db") + assert result == "postgresql+asyncpg://user@localhost/db" + + def test_unsupported_scheme_returns_original_uri(self): + """Test that unsupported schemes return the original URI unchanged.""" + uri = "oracle://user:pass@localhost:1521/dbname" + result = settings._get_async_conn_uri_from_sync(uri) + assert result == uri + + def test_empty_string_raises_value_error(self): + """Test that empty string raises ValueError with helpful message.""" + with pytest.raises(ValueError, match="Invalid SQLAlchemy connection URI"): + settings._get_async_conn_uri_from_sync("") + + def test_none_raises_value_error(self): + """Test that None raises ValueError with helpful message.""" + with pytest.raises(ValueError, match="Invalid SQLAlchemy connection URI"): + settings._get_async_conn_uri_from_sync(None) + + def test_malformed_uri_without_colon_raises_value_error(self): + """Test that URI without ':' separator raises ValueError.""" + with pytest.raises(ValueError, match="Invalid SQLAlchemy connection URI.*':' separator"): + settings._get_async_conn_uri_from_sync("notavaliduri") + + def test_error_message_is_helpful(self): + """Test that error message contains helpful guidance.""" + with pytest.raises(ValueError, match="Invalid SQLAlchemy connection URI") as exc_info: + settings._get_async_conn_uri_from_sync("invalid_value") + + error_msg = str(exc_info.value) + assert "AIRFLOW__DATABASE__SQL_ALCHEMY_CONN" in error_msg + assert "sql_alchemy_conn" in error_msg + assert "airflow.cfg" in error_msg From cb02568c04186f6d82e3dfcada00cc2e6514be09 Mon Sep 17 00:00:00 2001 From: Gayathri Srividya Rajavarapu Date: Tue, 14 Jul 2026 20:41:50 +0530 Subject: [PATCH 2/3] Parametrize settings URI conversion tests --- airflow-core/tests/unit/core/test_settings.py | 58 ++++++++----------- 1 file changed, 25 insertions(+), 33 deletions(-) diff --git a/airflow-core/tests/unit/core/test_settings.py b/airflow-core/tests/unit/core/test_settings.py index 126610f1ce944..479a153114bda 100644 --- a/airflow-core/tests/unit/core/test_settings.py +++ b/airflow-core/tests/unit/core/test_settings.py @@ -538,25 +538,19 @@ def test_early_return_when_all_none(self): class TestGetAsyncConnUriFromSync: """Tests for _get_async_conn_uri_from_sync function.""" - def test_sqlite_uri_conversion(self): - """Test conversion of SQLite sync URI to async with aiosqlite.""" - result = settings._get_async_conn_uri_from_sync("sqlite:///path/to/db.sqlite") - assert result == "sqlite+aiosqlite:///path/to/db.sqlite" - - def test_postgresql_uri_conversion(self): - """Test conversion of PostgreSQL sync URI to async with asyncpg.""" - result = settings._get_async_conn_uri_from_sync("postgresql://user:pass@localhost/dbname") - assert result == "postgresql+asyncpg://user:pass@localhost/dbname" - - def test_mysql_uri_conversion(self): - """Test conversion of MySQL sync URI to async with aiomysql.""" - result = settings._get_async_conn_uri_from_sync("mysql://user:pass@localhost/dbname") - assert result == "mysql+aiomysql://user:pass@localhost/dbname" - - def test_postgresql_psycopg2_uri_conversion(self): - """Test conversion of PostgreSQL with psycopg2 driver to asyncpg.""" - result = settings._get_async_conn_uri_from_sync("postgresql+psycopg2://user@localhost/db") - assert result == "postgresql+asyncpg://user@localhost/db" + @pytest.mark.parametrize( + ("sync_uri", "expected"), + [ + ("sqlite:///path/to/db.sqlite", "sqlite+aiosqlite:///path/to/db.sqlite"), + ("postgresql://user:pass@localhost/dbname", "postgresql+asyncpg://user:pass@localhost/dbname"), + ("mysql://user:pass@localhost/dbname", "mysql+aiomysql://user:pass@localhost/dbname"), + ("postgresql+psycopg2://user@localhost/db", "postgresql+asyncpg://user@localhost/db"), + ], + ) + def test_supported_scheme_conversion(self, sync_uri, expected): + """Test conversion of supported sync SQLAlchemy URIs to async driver variants.""" + result = settings._get_async_conn_uri_from_sync(sync_uri) + assert result == expected def test_unsupported_scheme_returns_original_uri(self): """Test that unsupported schemes return the original URI unchanged.""" @@ -564,20 +558,18 @@ def test_unsupported_scheme_returns_original_uri(self): result = settings._get_async_conn_uri_from_sync(uri) assert result == uri - def test_empty_string_raises_value_error(self): - """Test that empty string raises ValueError with helpful message.""" - with pytest.raises(ValueError, match="Invalid SQLAlchemy connection URI"): - settings._get_async_conn_uri_from_sync("") - - def test_none_raises_value_error(self): - """Test that None raises ValueError with helpful message.""" - with pytest.raises(ValueError, match="Invalid SQLAlchemy connection URI"): - settings._get_async_conn_uri_from_sync(None) - - def test_malformed_uri_without_colon_raises_value_error(self): - """Test that URI without ':' separator raises ValueError.""" - with pytest.raises(ValueError, match="Invalid SQLAlchemy connection URI.*':' separator"): - settings._get_async_conn_uri_from_sync("notavaliduri") + @pytest.mark.parametrize( + ("invalid_uri", "error_match"), + [ + ("", "Invalid SQLAlchemy connection URI"), + (None, "Invalid SQLAlchemy connection URI"), + ("notavaliduri", "Invalid SQLAlchemy connection URI.*':' separator"), + ], + ) + def test_invalid_uri_raises_value_error(self, invalid_uri, error_match): + """Test that invalid URIs raise ValueError with actionable guidance.""" + with pytest.raises(ValueError, match=error_match): + settings._get_async_conn_uri_from_sync(invalid_uri) def test_error_message_is_helpful(self): """Test that error message contains helpful guidance.""" From 3b53129093773cd63ed89f5d40b329a64a91b251 Mon Sep 17 00:00:00 2001 From: Gayathri Srividya Rajavarapu Date: Tue, 14 Jul 2026 20:59:23 +0530 Subject: [PATCH 3/3] Handle psycopg async driver in URI conversion tests --- airflow-core/tests/unit/core/test_settings.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/airflow-core/tests/unit/core/test_settings.py b/airflow-core/tests/unit/core/test_settings.py index 479a153114bda..7c0e48e825775 100644 --- a/airflow-core/tests/unit/core/test_settings.py +++ b/airflow-core/tests/unit/core/test_settings.py @@ -542,9 +542,7 @@ class TestGetAsyncConnUriFromSync: ("sync_uri", "expected"), [ ("sqlite:///path/to/db.sqlite", "sqlite+aiosqlite:///path/to/db.sqlite"), - ("postgresql://user:pass@localhost/dbname", "postgresql+asyncpg://user:pass@localhost/dbname"), ("mysql://user:pass@localhost/dbname", "mysql+aiomysql://user:pass@localhost/dbname"), - ("postgresql+psycopg2://user@localhost/db", "postgresql+asyncpg://user@localhost/db"), ], ) def test_supported_scheme_conversion(self, sync_uri, expected): @@ -552,6 +550,20 @@ def test_supported_scheme_conversion(self, sync_uri, expected): result = settings._get_async_conn_uri_from_sync(sync_uri) assert result == expected + @pytest.mark.parametrize( + ("sync_uri", "expected_template"), + [ + ("postgresql://user:pass@localhost/dbname", "postgresql+{driver}://user:pass@localhost/dbname"), + ("postgresql+psycopg2://user@localhost/db", "postgresql+{driver}://user@localhost/db"), + ], + ) + def test_postgresql_scheme_conversion(self, sync_uri, expected_template): + """Test conversion of PostgreSQL sync URIs to the configured async driver.""" + postgres_async_driver = "psycopg_async" if settings._USE_PSYCOPG3 else "asyncpg" + assert settings._get_async_conn_uri_from_sync(sync_uri) == expected_template.format( + driver=postgres_async_driver + ) + def test_unsupported_scheme_returns_original_uri(self): """Test that unsupported schemes return the original URI unchanged.""" uri = "oracle://user:pass@localhost:1521/dbname"