jason810496 commented on code in PR #72904:
URL: https://github.com/apache/airflow/pull/72904#discussion_r4005542559
##########
task-sdk/tests/task_sdk/test_providers_manager_runtime.py:
##########
@@ -112,8 +112,13 @@ def test_warning_logs_generated(self):
)
providers_manager._discover_hooks()
_ = providers_manager._hooks_lazy_dict["wrong-connection-type"]
- assert len(self._caplog.entries) == 1
- assert "Inconsistency!" in self._caplog[0]["event"]
+ # 'wrong-connection-type' is also read back under a different name, so
discovery
+ # warns about that as well. Both are expected, and no others.
+ assert len(self._caplog.entries) == 2
+ assert sum("Inconsistency!" in entry["event"] for entry in
self._caplog.entries) == 1
+ assert (
+ sum("read back under a different name" in entry["event"] for entry
in self._caplog.entries) == 1
+ )
Review Comment:
`len == 2` plus two independent substring-sum checks isn't actually
exhaustive: an entry matching both substrings plus an unvalidated third entry
would still satisfy all three. Adding an `all(...)` check that every entry
matches one of the two known patterns closes that gap.
```suggestion
assert len(self._caplog.entries) == 2
assert sum("Inconsistency!" in entry["event"] for entry in
self._caplog.entries) == 1
assert (
sum("read back under a different name" in entry["event"] for
entry in self._caplog.entries) == 1
)
assert all(
"Inconsistency!" in entry["event"] or "read back under a
different name" in entry["event"]
for entry in self._caplog.entries
)
```
##########
task-sdk/src/airflow/sdk/providers_manager_runtime.py:
##########
@@ -395,6 +395,80 @@ def _discover_hooks(self) -> None:
provider_uses_connection_types,
)
self._hook_provider_dict =
dict(sorted(self._hook_provider_dict.items()))
+ self._warn_about_unresolvable_connection_types()
+
+ @staticmethod
+ def _connection_type_as_stored(connection_type: str) -> str:
+ """
+ Return the connection type that a stored connection presents for this
declaration.
+
+ This follows what a connection goes through on the way out and back in.
+ ``Connection.get_uri()`` lowercases the type and encodes '_' as '-',
because RFC 3986
+ forbids '_' in a URI scheme, and reading a connection back from a URI
or from JSON
+ decodes it again. A declaration that does not come back unchanged is
registered under
+ a name no such connection ever presents.
+ """
+ # Imported here because airflow.sdk.definitions.connection imports
this module at
+ # module level. _normalize_conn_type is the decoder itself, so calling
it keeps this
+ # in step with the aliases it applies rather than restating them.
+ from airflow.sdk.definitions.connection import Connection
+
+ scheme = urlsplit(f"{connection_type.lower().replace('_',
'-')}://host").scheme
+ return Connection._normalize_conn_type(scheme)
+
+ def _warn_about_unresolvable_connection_types(self) -> None:
+ """
+ Warn about a declared connection type that a stored connection cannot
resolve.
+
+ A hook registers under the ``connection-type`` string verbatim, while
a connection
+ read from a URI, from JSON, or rebuilt from the secrets cache presents
the decoded
+ form of that string. Where the two differ, the hook is unreachable for
those
+ connections. A connection created directly through the UI, the REST
API or the CLI
+ keeps the declared spelling and does resolve, which is what lets the
mismatch go
+ unnoticed until a connection is served from somewhere else.
+
+ Where two providers declare two spellings of one name it is worse than
unreachable: a
+ connection for either resolves whichever hook holds the decoded name,
and because that
+ resolves rather than failing, there is no error anywhere to carry the
explanation.
+
+ Discovery is the only place either can be reported.
provider.yaml.schema.json
+ validates the providers in this repository alone, so a third-party
distribution never
+ passes through it, and Connection.get_hook() is reached only once a
lookup has already
+ failed, which the collision never does.
+ """
+ declared_by_stored_name: dict[str, list[str]] = {}
+ for connection_type in self._hook_provider_dict:
+ stored_name = self._connection_type_as_stored(connection_type)
+ declared_by_stored_name.setdefault(stored_name,
[]).append(connection_type)
+
+ for stored_name, declared in declared_by_stored_name.items():
+ if all(connection_type == stored_name for connection_type in
declared):
+ continue
Review Comment:
Dict keys are distinct, so "all equal `stored_name`" only ever holds when
there's exactly one. Stating the actual invariant directly is easier to trust
than making the reader re-derive it from key-uniqueness.
```suggestion
if len(declared) == 1 and declared[0] == stored_name:
continue
```
##########
task-sdk/src/airflow/sdk/providers_manager_runtime.py:
##########
@@ -395,6 +395,80 @@ def _discover_hooks(self) -> None:
provider_uses_connection_types,
)
self._hook_provider_dict =
dict(sorted(self._hook_provider_dict.items()))
+ self._warn_about_unresolvable_connection_types()
+
+ @staticmethod
+ def _connection_type_as_stored(connection_type: str) -> str:
+ """
+ Return the connection type that a stored connection presents for this
declaration.
+
+ This follows what a connection goes through on the way out and back in.
+ ``Connection.get_uri()`` lowercases the type and encodes '_' as '-',
because RFC 3986
+ forbids '_' in a URI scheme, and reading a connection back from a URI
or from JSON
+ decodes it again. A declaration that does not come back unchanged is
registered under
+ a name no such connection ever presents.
+ """
+ # Imported here because airflow.sdk.definitions.connection imports
this module at
+ # module level. _normalize_conn_type is the decoder itself, so calling
it keeps this
+ # in step with the aliases it applies rather than restating them.
+ from airflow.sdk.definitions.connection import Connection
+
+ scheme = urlsplit(f"{connection_type.lower().replace('_',
'-')}://host").scheme
+ return Connection._normalize_conn_type(scheme)
+
+ def _warn_about_unresolvable_connection_types(self) -> None:
+ """
+ Warn about a declared connection type that a stored connection cannot
resolve.
+
+ A hook registers under the ``connection-type`` string verbatim, while
a connection
+ read from a URI, from JSON, or rebuilt from the secrets cache presents
the decoded
+ form of that string. Where the two differ, the hook is unreachable for
those
+ connections. A connection created directly through the UI, the REST
API or the CLI
+ keeps the declared spelling and does resolve, which is what lets the
mismatch go
+ unnoticed until a connection is served from somewhere else.
+
+ Where two providers declare two spellings of one name it is worse than
unreachable: a
+ connection for either resolves whichever hook holds the decoded name,
and because that
+ resolves rather than failing, there is no error anywhere to carry the
explanation.
+
+ Discovery is the only place either can be reported.
provider.yaml.schema.json
+ validates the providers in this repository alone, so a third-party
distribution never
+ passes through it, and Connection.get_hook() is reached only once a
lookup has already
+ failed, which the collision never does.
+ """
+ declared_by_stored_name: dict[str, list[str]] = {}
+ for connection_type in self._hook_provider_dict:
+ stored_name = self._connection_type_as_stored(connection_type)
+ declared_by_stored_name.setdefault(stored_name,
[]).append(connection_type)
+
+ for stored_name, declared in declared_by_stored_name.items():
+ if all(connection_type == stored_name for connection_type in
declared):
+ continue
+ connection_types = sorted(declared)
+ packages = [self._hook_provider_dict[name].package_name for name
in connection_types]
+ if not stored_name:
+ log.warning(
+ "A declared connection type cannot be carried in a
connection URI, so a "
+ "connection read from a URI or from JSON has no connection
type at all.",
+ connection_types=connection_types,
+ packages=packages,
+ )
+ elif len(connection_types) > 1:
+ log.warning(
+ "Several declared connection types are read back under one
name, so a "
+ "connection for any of them resolves whichever hook holds
that name.",
+ connection_types=connection_types,
+ packages=packages,
+ read_back_as=stored_name,
+ )
+ else:
+ log.warning(
+ "A declared connection type is read back under a different
name, so a "
+ "connection read from a URI or from JSON cannot reach its
hook.",
+ connection_type=connection_types[0],
+ package=packages[0],
+ read_back_as=stored_name,
+ )
Review Comment:
This branch uses singular `connection_type`/`package` while the two branches
above use plural `connection_types`/`packages` for the same concept. A
log-based alert keyed on the plural fields (the more severe cases) would
silently never match this one. Suggest unifying the schema (needs a matching
test update at
`test_warns_about_a_connection_type_read_back_under_another_name`).
```suggestion
else:
log.warning(
"A declared connection type is read back under a
different name, so a "
"connection read from a URI or from JSON cannot reach
its hook.",
connection_types=connection_types,
packages=packages,
read_back_as=stored_name,
)
```
##########
task-sdk/tests/task_sdk/test_providers_manager_runtime.py:
##########
@@ -164,6 +169,133 @@ def test_already_registered_conn_type_in_provide(self):
" and 'airflow.providers.dummy.hooks.dummy.DummyHook2'."
) in msg
+ @staticmethod
+ def _provider_declaring(*connection_types: str) -> ProviderInfo:
+ return ProviderInfo(
+ version="0.0.1",
+ data={
+ "connection-types": [
+ {
+ "hook-class-name":
f"airflow.providers.dummy.hooks.dummy.Hook{index}",
+ "connection-type": connection_type,
+ }
+ for index, connection_type in enumerate(connection_types)
+ ],
+ },
+ )
+
+ @pytest.mark.parametrize(
+ ("declared", "read_back_as"),
+ [
+ # '-' is the URI-scheme encoding of '_', so it is decoded on the
way back in.
+ pytest.param("dummy-vendor", "dummy_vendor", id="hyphen"),
+ # get_uri() lowercases the scheme.
+ pytest.param("DummyVendor", "dummyvendor", id="uppercase"),
+ # _normalize_conn_type also applies this alias, which is why the
check asks it
+ # rather than restating the separator rule.
+ pytest.param("postgresql", "postgres", id="alias"),
+ ],
+ )
+ def test_warns_about_a_connection_type_read_back_under_another_name(self,
declared, read_back_as):
+ """
+ Such a type registers verbatim and resolves for a connection created
through the UI,
+ the REST API or the CLI, so nothing fails there. Every connection read
from a URI or
+ from JSON presents the decoded name instead and never reaches the hook.
+ """
+ with self._caplog.at_level(logging.WARNING):
+ providers_manager = ProvidersManagerTaskRuntime()
+ providers_manager._provider_dict["apache-airflow-providers-dummy"]
= self._provider_declaring(
+ declared
+ )
+ providers_manager._discover_hooks()
+
+ entries = [
+ entry for entry in self._caplog.entries if "read back under a
different name" in entry["event"]
+ ]
+ assert len(entries) == 1
+ assert entries[0]["connection_type"] == declared
+ assert entries[0]["read_back_as"] == read_back_as
+ assert entries[0]["package"] == "apache-airflow-providers-dummy"
Review Comment:
Update to match the field-name unification suggested above
(`connection_type`/`package` → `connection_types`/`packages`).
```suggestion
assert entries[0]["connection_types"] == [declared]
assert entries[0]["read_back_as"] == read_back_as
assert entries[0]["packages"] == ["apache-airflow-providers-dummy"]
```
##########
task-sdk/tests/task_sdk/test_providers_manager_runtime.py:
##########
@@ -164,6 +169,133 @@ def test_already_registered_conn_type_in_provide(self):
" and 'airflow.providers.dummy.hooks.dummy.DummyHook2'."
) in msg
+ @staticmethod
+ def _provider_declaring(*connection_types: str) -> ProviderInfo:
+ return ProviderInfo(
+ version="0.0.1",
+ data={
+ "connection-types": [
+ {
+ "hook-class-name":
f"airflow.providers.dummy.hooks.dummy.Hook{index}",
+ "connection-type": connection_type,
+ }
+ for index, connection_type in enumerate(connection_types)
+ ],
+ },
+ )
+
+ @pytest.mark.parametrize(
+ ("declared", "read_back_as"),
+ [
+ # '-' is the URI-scheme encoding of '_', so it is decoded on the
way back in.
+ pytest.param("dummy-vendor", "dummy_vendor", id="hyphen"),
+ # get_uri() lowercases the scheme.
+ pytest.param("DummyVendor", "dummyvendor", id="uppercase"),
+ # _normalize_conn_type also applies this alias, which is why the
check asks it
+ # rather than restating the separator rule.
+ pytest.param("postgresql", "postgres", id="alias"),
+ ],
+ )
+ def test_warns_about_a_connection_type_read_back_under_another_name(self,
declared, read_back_as):
+ """
+ Such a type registers verbatim and resolves for a connection created
through the UI,
+ the REST API or the CLI, so nothing fails there. Every connection read
from a URI or
+ from JSON presents the decoded name instead and never reaches the hook.
+ """
+ with self._caplog.at_level(logging.WARNING):
+ providers_manager = ProvidersManagerTaskRuntime()
+ providers_manager._provider_dict["apache-airflow-providers-dummy"]
= self._provider_declaring(
+ declared
+ )
+ providers_manager._discover_hooks()
+
+ entries = [
+ entry for entry in self._caplog.entries if "read back under a
different name" in entry["event"]
+ ]
+ assert len(entries) == 1
+ assert entries[0]["connection_type"] == declared
+ assert entries[0]["read_back_as"] == read_back_as
+ assert entries[0]["package"] == "apache-airflow-providers-dummy"
+
+ def test_warns_about_a_connection_type_a_uri_cannot_carry(self):
+ """A type that is not a usable scheme is lost altogether rather than
re-spelled."""
+ with self._caplog.at_level(logging.WARNING):
+ providers_manager = ProvidersManagerTaskRuntime()
+ providers_manager._provider_dict["apache-airflow-providers-dummy"]
= self._provider_declaring(
+ "dummy vendor"
+ )
+ providers_manager._discover_hooks()
+
+ entries = [
+ entry
+ for entry in self._caplog.entries
+ if "cannot be carried in a connection URI" in entry["event"]
+ ]
+ assert len(entries) == 1
+ assert entries[0]["connection_types"] == ["dummy vendor"]
+
+ def
test_warns_when_two_providers_declare_the_two_spellings_of_one_name(self):
+ """
+ This one resolves rather than failing, which is why get_hook() cannot
report it: the
+ connection is handed whichever hook holds the decoded name, so it can
belong to the
+ other provider.
+ """
+ with self._caplog.at_level(logging.WARNING):
+ providers_manager = ProvidersManagerTaskRuntime()
+ providers_manager._provider_dict["apache-airflow-providers-one"] =
self._provider_declaring(
+ "shared-name"
+ )
+ providers_manager._provider_dict["apache-airflow-providers-two"] =
self._provider_declaring(
+ "shared_name"
+ )
+ providers_manager._discover_hooks()
+
+ entries = [entry for entry in self._caplog.entries if "read back under
one name" in entry["event"]]
+ assert len(entries) == 1
+ assert entries[0]["connection_types"] == ["shared-name", "shared_name"]
+ assert entries[0]["read_back_as"] == "shared_name"
+ assert sorted(entries[0]["packages"]) == [
+ "apache-airflow-providers-one",
+ "apache-airflow-providers-two",
+ ]
+
+ def
test_does_not_warn_about_a_connection_type_that_survives_being_stored(self):
+ with self._caplog.at_level(logging.WARNING):
+ providers_manager = ProvidersManagerTaskRuntime()
+ providers_manager._provider_dict["apache-airflow-providers-dummy"]
= self._provider_declaring(
+ "dummy_vendor", "postgres", "s3", "a.b"
+ )
+ providers_manager._discover_hooks()
+
+ assert not self._caplog.entries
+
+ @pytest.mark.parametrize(
+ "declared",
+ [
+ "dummy_vendor",
+ "dummy-vendor",
+ "DummyVendor",
+ "postgres",
+ "postgresql",
+ "dummy vendor",
+ "a.b",
+ "a+b",
+ "foo-bar_baz",
+ ],
+ )
+ def test_stored_name_matches_a_real_connection_round_trip(self, declared):
+ """
+ The check models what get_uri() writes and what reading a connection
back decodes,
+ so it has to agree with actually doing it. This fails if either side
changes.
+ """
+ from airflow.sdk.definitions.connection import Connection
+
Review Comment:
Per this repo's convention, non-top-level imports are only for
circular-import avoidance, lazy worker-isolation loading, or `TYPE_CHECKING`
blocks. None apply here — this is a test file with no circular-import
relationship with `connection.py`. Suggest moving this to the file's top-level
imports instead.
```suggestion
```
##########
task-sdk/src/airflow/sdk/providers_manager_runtime.py:
##########
@@ -395,6 +395,80 @@ def _discover_hooks(self) -> None:
provider_uses_connection_types,
)
self._hook_provider_dict =
dict(sorted(self._hook_provider_dict.items()))
+ self._warn_about_unresolvable_connection_types()
+
+ @staticmethod
+ def _connection_type_as_stored(connection_type: str) -> str:
+ """
+ Return the connection type that a stored connection presents for this
declaration.
+
+ This follows what a connection goes through on the way out and back in.
+ ``Connection.get_uri()`` lowercases the type and encodes '_' as '-',
because RFC 3986
+ forbids '_' in a URI scheme, and reading a connection back from a URI
or from JSON
+ decodes it again. A declaration that does not come back unchanged is
registered under
+ a name no such connection ever presents.
+ """
+ # Imported here because airflow.sdk.definitions.connection imports
this module at
+ # module level. _normalize_conn_type is the decoder itself, so calling
it keeps this
+ # in step with the aliases it applies rather than restating them.
+ from airflow.sdk.definitions.connection import Connection
+
+ scheme = urlsplit(f"{connection_type.lower().replace('_',
'-')}://host").scheme
+ return Connection._normalize_conn_type(scheme)
+
+ def _warn_about_unresolvable_connection_types(self) -> None:
+ """
+ Warn about a declared connection type that a stored connection cannot
resolve.
+
+ A hook registers under the ``connection-type`` string verbatim, while
a connection
+ read from a URI, from JSON, or rebuilt from the secrets cache presents
the decoded
+ form of that string. Where the two differ, the hook is unreachable for
those
+ connections. A connection created directly through the UI, the REST
API or the CLI
+ keeps the declared spelling and does resolve, which is what lets the
mismatch go
+ unnoticed until a connection is served from somewhere else.
+
+ Where two providers declare two spellings of one name it is worse than
unreachable: a
+ connection for either resolves whichever hook holds the decoded name,
and because that
+ resolves rather than failing, there is no error anywhere to carry the
explanation.
+
+ Discovery is the only place either can be reported.
provider.yaml.schema.json
+ validates the providers in this repository alone, so a third-party
distribution never
+ passes through it, and Connection.get_hook() is reached only once a
lookup has already
+ failed, which the collision never does.
+ """
+ declared_by_stored_name: dict[str, list[str]] = {}
+ for connection_type in self._hook_provider_dict:
+ stored_name = self._connection_type_as_stored(connection_type)
+ declared_by_stored_name.setdefault(stored_name,
[]).append(connection_type)
+
+ for stored_name, declared in declared_by_stored_name.items():
+ if all(connection_type == stored_name for connection_type in
declared):
+ continue
+ connection_types = sorted(declared)
Review Comment:
`self._hook_provider_dict` is already sorted by key a few lines above
(`dict(sorted(self._hook_provider_dict.items()))`), so iterating it and
grouping into `declared` already yields sorted order — this re-sort is
redundant.
```suggestion
connection_types = declared
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
