kaxil commented on code in PR #72904:
URL: https://github.com/apache/airflow/pull/72904#discussion_r4008574994
##########
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:
Done. Clearer stated directly than left for the reader to derive from key
uniqueness.
##########
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:
Keeping this one. It is redundant today, but the two multi-spelling messages
have their field order asserted in tests, and dropping the sort would make that
depend on the `dict(sorted(...))` up in `_discover_hooks` rather than on
anything local to this method. It is a one or two element list, so keeping the
guarantee where it is relied on costs nothing.
It did turn up a `sorted()` worth removing though, a different one:
`sorted(entries[0]["packages"])` in
`test_warns_when_two_providers_declare_the_two_spellings_of_one_name`.
`packages` is built positionally from `connection_types`, which the same test
pins in exact order two lines above, so which provider declared which spelling
is the actual claim, and sorting it discarded exactly that. Reverse the two
packages and the assertion still passed. Fixed in this push.
##########
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:
Done. The reason these warnings pass keyword arguments at all is that a
positional argument is interpolated into the message and then dropped, while
keyword arguments stay as separate fields in the JSON event. One field schema
across the three events is that same argument carried through, so this is
consistent. Singular read better in a rendered line, but not at the price of a
plural-keyed query missing the single case.
##########
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:
Added. The gap it closes is one entry matching both substrings while the
other matches neither, which satisfies all three counts and leaves an entry
unchecked.
##########
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:
Done, following the rename above.
##########
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:
Hoisted. The inline import in the production method is a real cycle,
`airflow/sdk/definitions/connection.py:29` imports
`ProvidersManagerTaskRuntime` at module level, but the test module is not part
of that, so there was no reason for it there.
--
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]
