You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Dev-iL (via GitHub)" <gi...@apache.org> on 2026/09/20 18:21:49 UTC
[PR] Make Execution API secret reads async [airflow]
Dev-iL opened a new pull request, #73414:
URL: https://github.com/apache/airflow/pull/73414
related: #67799, #72329
Execution API `Variable` and `Connection` reads previously depended on synchronous core resolvers, preventing the server from using native async backend I/O. This change adds async server-side resolution while preserving the existing secrets backend configuration, ordering, and behavior:
- Adds optional `aget_variable` and `aget_connection` dispatch.
- Uses native async SQLAlchemy reads for the metastore backend.
- Runs unchanged synchronous backends, cache access, masking, and connection processing in bounded worker threads.
- Preserves custom sync overrides, team scope, cache semantics, access-denial behavior, cancellation, and existing HTTP responses.
- Migrates the Variable and Connection GET routes to await the new resolver.
- Disposes the test async engine inside each `TestClient` event loop, preventing pooled MySQL connections from crossing loop lifetimes.
- Documents async backend adoption, fallback precedence, resource ownership, cancellation limits, and rollback.
Native cloud-provider clients remain follow-up work, separate from the SDK Variable API work in #72329.
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [x] Yes (please specify the tool below)
Generated-by: Codex following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4324887738
##########
airflow-core/src/airflow/secrets/resolver.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import contextlib
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.models.crypto import get_fernet
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+ from airflow.secrets import BaseSecretsBackend
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
Review Comment:
Fixed by moving the cache implementation to `shared/secrets_backend`, with no core or SDK imports or new dependencies. Core uses a real `airflow.secrets.cache` adapter configured by core; the SDK keeps its existing import paths and no-argument `init()` through its own adapter. The resolver's SDK import and suppression are removed, and synchronous core reads/writes use the core cache. The parsing bootstrap shares the existing manager dictionary and TTL between the adapters so SDK lookups and the in-process API retain coherent invalidation.
SKILLs follow-up: https://github.com/apache/airflow/pull/75212
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4324361637
##########
airflow-core/src/airflow/secrets/metastore.py:
##########
@@ -61,6 +90,36 @@ def get_connection(
session.expunge(conn)
return conn
+ async def aget_connection(
+ self, conn_id: str, team_name: str | None = None, *, session: AsyncSession | None = None
+ ) -> Connection | None:
+ """Get an Airflow Connection from the metadata DB using native async I/O."""
+ if session is None:
+ async with create_session_async() as owned_session:
+ return await self.aget_connection(conn_id, team_name=team_name, session=owned_session)
+
+ from airflow.models import Connection
+
+ row = (
+ (
+ await session.execute(
+ select(Connection.__table__)
Review Comment:
`select(Connection)` hydrates the ORM object and invokes [`Connection.on_db_load()`](https://github.com/apache/airflow/blob/e23a472a3321505f89d06efa175bccee0a634c3b/airflow-core/src/airflow/models/connection.py#L222) before the async query returns. That hook reads `password`, which calls [`get_fernet()`](https://github.com/apache/airflow/blob/e23a472a3321505f89d06efa175bccee0a634c3b/airflow-core/src/airflow/models/crypto.py). If initialization failed, this retries the key lookup on the event loop; configuration can invoke a command or a synchronous secrets backend. Expunging afterward is too late. SQLAlchemy documents the [load-time reconstructor behavior](https://docs.sqlalchemy.org/en/20/orm/mapping_api.html#sqlalchemy.orm.reconstructor).
The [column select](https://github.com/apache/airflow/blob/e23a472a3321505f89d06efa175bccee0a634c3b/airflow-core/src/airflow/secrets/metastore.py#L104) keeps database I/O on the borrowed async session and lets reconstruction run in a worker while the key is uncached. The session stays on its owning thread. The existing [`test_fernet_key_loads_stay_off_event_loop`](https://github.com/apache/airflow/blob/e23a472a3321505f89d06efa175bccee0a634c3b/airflow-core/tests/unit/always/test_secrets_resolver.py#L471) covers both successful initialization and repeated failure. I recommend retaining this split.
I checked a local async `select(Connection)` plus `expunge()` alternative against that regression: the valid-key case passes, but the failed-key case fails the assertion that every Fernet configuration read happens off the event-loop thread. The current column loader passes both.
---
Drafted-by: Codex (GPT-6) (reviewed by @Dev-iL before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "github-actions[bot] (via GitHub) via commits" <co...@airflow.apache.org>.
github-actions[bot] commented on PR #73414:
URL: https://github.com/apache/airflow/pull/73414#issuecomment-6631373363
<!-- fork-ci-gate: {"gate_checked": true, "sha": "80f3aac5bb4e76416aae4c09ed261bf6c218e12e", "since": "2026-11-04T13:03:07.655315+00:00", "state": "running"} -->
**Fork CI: running.** CI for the latest commit is running in your fork.
Pull requests from contributors run their CI in the contributor's own fork. GitHub Actions is free and has no minute limit on public forks, while the CI capacity of the Apache Software Foundation is shared by every ASF project and is finite.
Latest run: https://github.com/Dev-iL/airflow/actions/runs/41831794559
How this works: [fork CI for external contributors](https://github.com/apache/airflow/blob/main/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md).
---
_This comment was posted automatically by the fork CI monitor from templates drafted with an AI-assisted tool, so it may contain mistakes. An Apache Airflow maintainer, a real person, will review the pull request once its fork CI is green. [Why the first pass is automated](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated)._
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4323143302
##########
airflow-core/docs/security/secrets/secrets-backend/index.rst:
##########
@@ -161,6 +161,72 @@ section of ``airflow.cfg``.
Additional arguments to your SecretsBackend can be configured in ``airflow.cfg`` by supplying a JSON string to ``backend_kwargs``, which will be passed to the ``__init__`` of your SecretsBackend.
See :ref:`Configuration <secrets_backend_configuration>` for more details, and :ref:`SSM Parameter Store <ssm_parameter_store_secrets>` for an example.
+Async server-side reads
+^^^^^^^^^^^^^^^^^^^^^^^
+
+When the Execution API serves a Variable or Connection, it resolves the value with
+``airflow.secrets.resolver.resolve_variable`` or ``airflow.secrets.resolver.resolve_connection``.
+Both functions search the configured backends in the same order as synchronous reads. There is no
+separate async configuration.
+
+For backends configured on the Execution API server, either or both of these optional methods
+enable native async reads:
+
+.. code-block:: python
+
+ async def aget_variable(self, key: str, team_name: str | None = None) -> str | None: ...
+
+
+ async def aget_connection(self, conn_id: str, team_name: str | None = None) -> Connection | None: ...
+
+They take the same arguments and return the same values as ``get_variable`` and ``get_connection``.
+``aget_connection`` returns the core :py:class:`airflow.models.connection.Connection` or ``None``.
+
+The Execution API server's resolver calls these methods as follows. These precedence and
+``team_name`` forwarding rules apply to server-side resolution:
+
+* **Precedence:** the resolver uses the async method when it is defined in the same class as its
+ synchronous counterpart, or in a subclass of that class. If a subclass overrides only the
+ synchronous method, the resolver uses that override. Existing custom backends therefore keep their
+ behavior when a parent class gains an async method.
+* **Fallback:** a backend without an async method needs no change. The resolver runs its synchronous
+ method in a worker thread.
+* **Errors:** the resolver doesn't retry a failing async method synchronously. It continues with the
+ next backend, as it does for synchronous methods. ``AirflowSecretsBackendAccessDenied`` ends the
+ search.
+* **Event loop:** async methods run on the event loop that serves the request. Create, use, and close
+ async clients on that loop. Don't create a loop-bound client in ``__init__``, and don't share it
+ with the synchronous method.
+* **Database session:** the metastore backend's async methods use the request's async database
+ session when the Execution API supplies one. Otherwise, they open and close their own. The resolver
+ never passes that session to another backend, and it doesn't commit or close it.
+* **Cancellation:** a cancelled request stops the backend search and the cache update. The resolver
+ can't interrupt a synchronous method that is already running in a worker thread. That method
Review Comment:
It cant? Missing ability in to_thread, or too hard to wire up?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on PR #73414:
URL: https://github.com/apache/airflow/pull/73414#issuecomment-6248114456
(Marked as draft because I need the active PR slot for the 3.15 work)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4183577776
##########
airflow-core/src/airflow/secrets/resolver.py:
##########
Review Comment:
Renamed it to `airflow/secrets/resolver.py`, with `resolve_variable` and `resolve_connection`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4323358109
##########
airflow-core/src/airflow/secrets/resolver.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import contextlib
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.models.crypto import get_fernet
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+ from airflow.secrets import BaseSecretsBackend
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
Review Comment:
Regardless of the fix - I'll make sure this comment is present in the appropriate skill.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4183578843
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
+ raise ValueError(
+ "Multi-team mode is not configured in the Airflow environment but the task trying "
+ f"to access the {resource_name} belongs to a team"
+ )
+
+
+async def resolve_variable(
+ key: str, team_name: str | None = None, *, session: AsyncSession | None = None
+) -> str:
+ """Resolve a Variable without blocking the caller's event loop."""
+ await _validate_team_name(team_name, "variable")
+
+ SecretCache = _get_secret_cache()
+
+ try:
+ value = await run_in_threadpool(SecretCache.get_variable, key, team_name=team_name)
+ except SecretCache.NotPresentException:
+ pass
+ else:
+ if value is None:
+ raise KeyError(f"Variable {key} does not exist.")
+ await run_in_threadpool(mask_secret, value, key)
Review Comment:
Sure. Masking now runs directly on cache hits and backend results. Connection parsing, `get_uri()` and response validation also run directly. Fernet initialization remains in a worker because `core.fernet_key` can invoke `fernet_key_cmd` or query a secrets backend.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4185521246
##########
airflow-core/tests/unit/api_fastapi/execution_api/test_app.py:
##########
@@ -355,16 +446,24 @@ def _build_app(mode: str):
[
pytest.param("unsafe-always", "/health", False, True, 200, id="always-unauthenticated"),
pytest.param("unsafe-always", "/variables/k", False, True, 401, id="always-auth-failure"),
- pytest.param("unsafe-always", "/variables/k", True, True, None, id="always-authenticated"),
+ pytest.param("unsafe-always", "/variables/k", True, True, 200, id="always-authenticated"),
pytest.param("only-authenticated", "/health", False, False, 200, id="onlyauth-unauthenticated"),
pytest.param("only-authenticated", "/variables/k", False, False, 401, id="onlyauth-auth-failure"),
- pytest.param("only-authenticated", "/variables/k", True, True, None, id="onlyauth-authenticated"),
+ pytest.param("only-authenticated", "/variables/k", True, True, 200, id="onlyauth-authenticated"),
pytest.param("never", "/health", False, False, 200, id="never-unauthenticated"),
pytest.param("never", "/variables/k", False, False, 401, id="never-auth-failure"),
- pytest.param("never", "/variables/k", True, False, None, id="never-authenticated"),
+ pytest.param("never", "/variables/k", True, False, 200, id="never-authenticated"),
Review Comment:
This doesn't change the return codes of endpoints.
Previously, `expect_status=None` meant “skip the status assertion”. The test now mocks the resolver to return `"value"`, making successful requests deterministically return `200`.
This adapts the trace-context test to async resolution and keeps it from opening database connections with an unmanaged engine lifetime. It **doesn’t fix an existing endpoint-status bug**.
The route contract remains unchanged: success `200`, unauthenticated `401`, denied `403`, missing Variable `404`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4183578123
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
Review Comment:
`conf.getboolean("core", "multi_team")` now runs directly. That option isn't sensitive, so it doesn't invoke command or secret configuration lookups.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4323192510
##########
airflow-core/src/airflow/secrets/resolver.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import contextlib
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.models.crypto import get_fernet
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+ from airflow.secrets import BaseSecretsBackend
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
Review Comment:
New things in core should not be importing from sdk
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4323172919
##########
airflow-core/src/airflow/secrets/metastore.py:
##########
@@ -61,6 +90,36 @@ def get_connection(
session.expunge(conn)
return conn
+ async def aget_connection(
+ self, conn_id: str, team_name: str | None = None, *, session: AsyncSession | None = None
+ ) -> Connection | None:
+ """Get an Airflow Connection from the metadata DB using native async I/O."""
+ if session is None:
+ async with create_session_async() as owned_session:
+ return await self.aget_connection(conn_id, team_name=team_name, session=owned_session)
+
+ from airflow.models import Connection
+
+ row = (
+ (
+ await session.execute(
+ select(Connection.__table__)
Review Comment:
Why this, not `select(Connection)`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4183577776
##########
airflow-core/src/airflow/secrets/resolver.py:
##########
Review Comment:
Renamed it to `airflow/secrets/resolver.py`, with `resolve_variable` and `resolve_connection`. Imports, patch targets, tests and docs use the new name: [resolver.py](https://github.com/apache/airflow/blob/d69564c1a62fa51ad0d3a3450ede7aaed9c9750f/airflow-core/src/airflow/secrets/resolver.py).
---
Drafted-by: Codex (GPT-6) (no human review before posting)
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
Review Comment:
Agreed. `conf.getboolean("core", "multi_team")` now runs directly. That option isn't sensitive, so it doesn't invoke command or secret configuration lookups.
The remaining thread hops cover synchronous backend calls, backend-list loading that can import and construct custom backends, initialized cache communication with its manager process, and Fernet initialization. Successful Fernet initialization is cached; failed loads can retry. The boundaries are in [resolver.py](https://github.com/apache/airflow/blob/d69564c1a62fa51ad0d3a3450ede7aaed9c9750f/airflow-core/src/airflow/secrets/resolver.py).
---
Drafted-by: Codex (GPT-6) (no human review before posting)
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
+ raise ValueError(
+ "Multi-team mode is not configured in the Airflow environment but the task trying "
+ f"to access the {resource_name} belongs to a team"
+ )
+
+
+async def resolve_variable(
+ key: str, team_name: str | None = None, *, session: AsyncSession | None = None
+) -> str:
+ """Resolve a Variable without blocking the caller's event loop."""
+ await _validate_team_name(team_name, "variable")
+
+ SecretCache = _get_secret_cache()
+
+ try:
+ value = await run_in_threadpool(SecretCache.get_variable, key, team_name=team_name)
Review Comment:
An initialized cache uses `multiprocessing.Manager().dict()`, so reads and writes communicate with the manager process. The resolver now calls an uninitialized cache directly, where calls only miss or do nothing, and uses a worker for the initialized cache.
`_call_secret_cache` checks `SecretCache._cache` because there is no public initialization predicate. The tests cover which thread each path uses: [resolver tests](https://github.com/apache/airflow/blob/d69564c1a62fa51ad0d3a3450ede7aaed9c9750f/airflow-core/tests/unit/always/test_secrets_resolver.py). The SDK cache API is unchanged.
---
Drafted-by: Codex (GPT-6) (no human review before posting)
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
+ raise ValueError(
+ "Multi-team mode is not configured in the Airflow environment but the task trying "
+ f"to access the {resource_name} belongs to a team"
+ )
+
+
+async def resolve_variable(
+ key: str, team_name: str | None = None, *, session: AsyncSession | None = None
+) -> str:
+ """Resolve a Variable without blocking the caller's event loop."""
+ await _validate_team_name(team_name, "variable")
+
+ SecretCache = _get_secret_cache()
+
+ try:
+ value = await run_in_threadpool(SecretCache.get_variable, key, team_name=team_name)
+ except SecretCache.NotPresentException:
+ pass
+ else:
+ if value is None:
+ raise KeyError(f"Variable {key} does not exist.")
+ await run_in_threadpool(mask_secret, value, key)
Review Comment:
Agreed. Masking now runs directly on cache hits and backend results. Connection parsing, `get_uri()` and response validation also run directly.
Fernet initialization remains in a worker because `core.fernet_key` can invoke `fernet_key_cmd` or query a secrets backend. Successful initialization is cached; failed loads can retry. The initialization boundary is [resolver.py](https://github.com/apache/airflow/blob/d69564c1a62fa51ad0d3a3450ede7aaed9c9750f/airflow-core/src/airflow/secrets/resolver.py).
---
Drafted-by: Codex (GPT-6) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4308098734
##########
airflow-core/docs/security/secrets/secrets-backend/index.rst:
##########
@@ -161,6 +161,62 @@ section of ``airflow.cfg``.
Additional arguments to your SecretsBackend can be configured in ``airflow.cfg`` by supplying a JSON string to ``backend_kwargs``, which will be passed to the ``__init__`` of your SecretsBackend.
See :ref:`Configuration <secrets_backend_configuration>` for more details, and :ref:`SSM Parameter Store <ssm_parameter_store_secrets>` for an example.
+Async server-side reads
+^^^^^^^^^^^^^^^^^^^^^^^
+
+When the Execution API serves a Variable or Connection, it resolves the value with
+``airflow.secrets.resolver.resolve_variable`` or ``airflow.secrets.resolver.resolve_connection``.
+Both functions search the configured backends in the same order as synchronous reads. There is no
+separate async configuration.
+
+A backend can implement either or both of these optional methods:
+
+.. code-block:: python
+
+ async def aget_variable(self, key: str, team_name: str | None = None) -> str | None: ...
+
+
+ async def aget_connection(self, conn_id: str, team_name: str | None = None) -> Connection | None: ...
+
+They take the same arguments and return the same values as ``get_variable`` and ``get_connection``.
+``aget_connection`` returns the core :py:class:`airflow.models.connection.Connection` or ``None``.
+
+The resolver calls these methods as follows:
+
+* **Precedence:** the resolver uses the async method when it is defined in the same class as its
Review Comment:
Scoped the contract explicitly to backends configured on the Execution API server. The page now also describes local worker lookups: environment/worker backends precede ExecutionAPISecretsBackend, the SDK prefers an available async method even over a subclass's sync override, and those local calls do not receive team_name. The SDK behavior is unchanged.
---
Drafted-by: Codex (GPT-6) (no human review before posting)
##########
airflow-core/src/airflow/secrets/metastore.py:
##########
@@ -23,15 +23,40 @@
from sqlalchemy import or_, select
+from airflow._shared.secrets_masker import mask_secret
from airflow.secrets import BaseSecretsBackend
-from airflow.utils.session import NEW_SESSION, provide_session
+from airflow.utils.session import NEW_SESSION, create_session_async, provide_session
if TYPE_CHECKING:
+ from sqlalchemy.engine import RowMapping
+ from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session
from airflow.models import Connection
+def _build_connection(values: RowMapping) -> Connection:
Review Comment:
There is a loop-safety constraint here: the ORM reconstructors decrypt during async query materialization, before expunge could help. Reading columns lets the key-dependent reconstruction run in a worker while Fernet is uncached, including after a failed key load, without moving the borrowed async session across threads. I've documented that beside the helpers. The metastore tests also bound the four reads to four queries.
---
Drafted-by: Codex (GPT-6) (no human review before posting)
##########
airflow-core/src/airflow/secrets/resolver.py:
##########
@@ -0,0 +1,241 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import contextlib
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.models.crypto import get_fernet
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+ from airflow.secrets import BaseSecretsBackend
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _call_secret_cache(method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+ # ``SecretCache.init()`` stores entries in a ``multiprocessing.Manager`` dict, so every access to an
+ # initialized cache is IPC to the manager process. An uninitialized cache only misses or no-ops.
+ if _get_secret_cache()._cache is None:
+ return method(*args, **kwargs)
+ return await run_in_threadpool(method, *args, **kwargs)
+
+
+async def _load_fernet() -> None:
+ # The first ``get_fernet()`` reads ``core.fernet_key``, which may run ``fernet_key_cmd`` or query a
+ # secrets backend. Once loaded, encrypting and decrypting stored values is local work. A failed
+ # load is not cached, so its error still surfaces where a value needs the key, as before.
+ if not get_fernet.cache_info().currsize:
+ with contextlib.suppress(Exception):
Review Comment:
Fixed the server-owned retry paths. While Fernet is uncached, cached-URI Connection construction, metastore reconstruction/decryption, and get_uri run in a worker. The initial warm-up remains, successful loads stay cached, and plaintext backend resolution still works if loading the key fails. Parameterized regressions exercise all four paths with valid and failing key loads, two consecutive requests, and assertions on the thread of every FERNET_KEY read. The four failure cases reproduce loop-thread reads before the fix.
---
Drafted-by: Codex (GPT-6) (no human review before posting)
##########
airflow-core/tests/unit/api_fastapi/execution_api/test_app.py:
##########
@@ -96,6 +97,96 @@ def test_server_and_in_process_apps_isolate_auth(in_process_first, monkeypatch):
assert authenticated_response.json() == response.json()
+@pytest.mark.parametrize("failure", [None, "startup", "shutdown", "dispose", "configure"])
+@mock.patch("airflow.api_fastapi.execution_api.app.InProcessExecutionAPI", autospec=True)
+def test_in_process_fixture_owns_engine_lifetime(mock_api, request, failure):
+ previous_engine, previous_factory = settings.async_engine, settings.AsyncSession
+ api = InProcessExecutionAPI()
+ mock_api.return_value = api
+ original_lifespan = api.app.router.lifespan_context
+ disposed = []
+ original_dispose = AsyncEngine.dispose
+ original_configure = settings._configure_async_session
+
+ @asynccontextmanager
+ async def lifespan(app):
+ async with original_lifespan(app):
+ if failure == "startup":
+ raise RuntimeError("startup failed")
+ yield
+ if failure == "shutdown":
+ raise RuntimeError("shutdown failed")
+
+ api.app.router.lifespan_context = lifespan
+
+ async def dispose(engine, *args, **kwargs):
+ disposed.append((engine, asyncio.get_running_loop()))
+ await original_dispose(engine, *args, **kwargs)
+ if engine is not previous_engine and failure == "dispose":
+ raise RuntimeError("dispose failed")
+
+ def configure():
+ original_configure()
+ if failure == "configure":
+ raise RuntimeError("configure failed")
+
+ async def query(factory):
+ async with factory() as session:
+ return await session.scalar(text("SELECT 1"))
+
+ @api.app.get("/fixture-db-read")
+ async def read():
+ return await query(settings.AsyncSession)
+
+ fixture = request._fixturemanager.getfixturedefs("in_process_execution_api", request.node)[-1].func
Review Comment:
Removed both fixture-introspection tests, including request._fixturemanager/getfixturedefs and fresh_async_engine.__wrapped__. The fixtures are still consumed through normal pytest injection, and the existing Dag processor and triggerer integration tests retain coverage of their use.
---
Drafted-by: Codex (GPT-6) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4307710100
##########
airflow-core/docs/security/secrets/secrets-backend/index.rst:
##########
@@ -161,6 +161,62 @@ section of ``airflow.cfg``.
Additional arguments to your SecretsBackend can be configured in ``airflow.cfg`` by supplying a JSON string to ``backend_kwargs``, which will be passed to the ``__init__`` of your SecretsBackend.
See :ref:`Configuration <secrets_backend_configuration>` for more details, and :ref:`SSM Parameter Store <ssm_parameter_store_secrets>` for an example.
+Async server-side reads
+^^^^^^^^^^^^^^^^^^^^^^^
+
+When the Execution API serves a Variable or Connection, it resolves the value with
+``airflow.secrets.resolver.resolve_variable`` or ``airflow.secrets.resolver.resolve_connection``.
+Both functions search the configured backends in the same order as synchronous reads. There is no
+separate async configuration.
+
+A backend can implement either or both of these optional methods:
+
+.. code-block:: python
+
+ async def aget_variable(self, key: str, team_name: str | None = None) -> str | None: ...
+
+
+ async def aget_connection(self, conn_id: str, team_name: str | None = None) -> Connection | None: ...
+
+They take the same arguments and return the same values as ``get_variable`` and ``get_connection``.
+``aget_connection`` returns the core :py:class:`airflow.models.connection.Connection` or ``None``.
+
+The resolver calls these methods as follows:
+
+* **Precedence:** the resolver uses the async method when it is defined in the same class as its
Review Comment:
**[warning]** This page states the `aget_*` contract for any backend, but the precedence rule (and `team_name` forwarding) only holds on the API server.
This section sits on the general secrets-backend page, so a backend author will read it as the contract for `aget_variable` / `aget_connection` wherever the backend is configured. In a task process the same methods are already picked up by the Task SDK with a different rule. [`_async_get_connection`](https://github.com/apache/airflow/blob/9ad8d78c27779b3cc6660d1a496676a4a195c203/task-sdk/src/airflow/sdk/execution_time/context.py#L294) and [`_async_get_variable`](https://github.com/apache/airflow/blob/9ad8d78c27779b3cc6660d1a496676a4a195c203/task-sdk/src/airflow/sdk/execution_time/context.py#L421) take `getattr(backend, "aget_*")` whenever it exists and call it without `team_name`. So on a worker:
- a subclass that overrides only `get_variable` still gets the inherited `aget_variable` (the reverse of the **Precedence** bullet);
- a team-scoped lookup never reaches `aget_*` with its team.
The last paragraph has the same gap: "Task processes read secrets through the Execution API" is true for the `ExecutionAPISecretsBackend`, but env vars and `[workers] secrets_backend` are read in the task process first, by the SDK path above.
Either scope the section to the Execution API server and say what the worker side does today, or align the SDK lookups with `_get_native_async_method` (here or in a follow-up linked from the page). One rule for one method name is easier to document than two.
##########
airflow-core/src/airflow/secrets/resolver.py:
##########
@@ -0,0 +1,241 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import contextlib
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.models.crypto import get_fernet
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+ from airflow.secrets import BaseSecretsBackend
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _call_secret_cache(method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+ # ``SecretCache.init()`` stores entries in a ``multiprocessing.Manager`` dict, so every access to an
+ # initialized cache is IPC to the manager process. An uninitialized cache only misses or no-ops.
+ if _get_secret_cache()._cache is None:
+ return method(*args, **kwargs)
+ return await run_in_threadpool(method, *args, **kwargs)
+
+
+async def _load_fernet() -> None:
+ # The first ``get_fernet()`` reads ``core.fernet_key``, which may run ``fernet_key_cmd`` or query a
+ # secrets backend. Once loaded, encrypting and decrypting stored values is local work. A failed
+ # load is not cached, so its error still surfaces where a value needs the key, as before.
+ if not get_fernet.cache_info().currsize:
+ with contextlib.suppress(Exception):
Review Comment:
**[warning]** When the Fernet load fails, the error is suppressed here and the load runs again on the event loop.
`get_fernet` is `@functools.cache`d, and `cache` stores no result for a call that raised. So after this `suppress`, the next `get_fernet()` re-reads `core.fernet_key` (and runs `fernet_key_cmd` / `fernet_key_secret`) on the loop thread. In this PR that call comes from:
- `Connection(conn_id=..., uri=...)` on a cache hit (via `set_password`);
- `on_db_load()` / `Variable.val` in the metastore `aget_*` methods;
- `connection.get_uri()` after a sync backend returns.
Probe on this head: `SecretCache` initialised with one cached connection URI with a password, `get_fernet` cache cleared, and `conf.get("core", "FERNET_KEY")` raising `AirflowConfigException`. Then `await resolve_connection("c")` reads the key twice: once in `AnyIO worker thread`, then on `MainThread` (the loop). Because failures are not cached, every request repeats both reads. When `fernet_key_cmd` is slow or hangs (a Vault CLI timing out), every route on that server stalls on each Connection or encrypted Variable read, task heartbeats included.
The comment above and the docs bullet ("Successful initialization is cached; failed loads can be retried") describe the retry but not where it runs. Two ways out, either keeps the loop free:
- re-raise the load error for the paths that need the key, so it surfaces once, from the worker;
- or run the decrypting step (`Connection(uri=...)`, `_build_connection`, `_get_variable_value`, `get_uri()`) in the worker while `get_fernet.cache_info().currsize == 0`.
A variant of `test_failed_fernet_key_load_does_not_stop_resolution` with an encrypted value, asserting the threads of the `FERNET_KEY` reads as `test_cold_fernet_key_is_loaded_off_the_event_loop_once` does, would cover it.
##########
airflow-core/tests/unit/api_fastapi/execution_api/test_app.py:
##########
@@ -96,6 +97,96 @@ def test_server_and_in_process_apps_isolate_auth(in_process_first, monkeypatch):
assert authenticated_response.json() == response.json()
+@pytest.mark.parametrize("failure", [None, "startup", "shutdown", "dispose", "configure"])
+@mock.patch("airflow.api_fastapi.execution_api.app.InProcessExecutionAPI", autospec=True)
+def test_in_process_fixture_owns_engine_lifetime(mock_api, request, failure):
+ previous_engine, previous_factory = settings.async_engine, settings.AsyncSession
+ api = InProcessExecutionAPI()
+ mock_api.return_value = api
+ original_lifespan = api.app.router.lifespan_context
+ disposed = []
+ original_dispose = AsyncEngine.dispose
+ original_configure = settings._configure_async_session
+
+ @asynccontextmanager
+ async def lifespan(app):
+ async with original_lifespan(app):
+ if failure == "startup":
+ raise RuntimeError("startup failed")
+ yield
+ if failure == "shutdown":
+ raise RuntimeError("shutdown failed")
+
+ api.app.router.lifespan_context = lifespan
+
+ async def dispose(engine, *args, **kwargs):
+ disposed.append((engine, asyncio.get_running_loop()))
+ await original_dispose(engine, *args, **kwargs)
+ if engine is not previous_engine and failure == "dispose":
+ raise RuntimeError("dispose failed")
+
+ def configure():
+ original_configure()
+ if failure == "configure":
+ raise RuntimeError("configure failed")
+
+ async def query(factory):
+ async with factory() as session:
+ return await session.scalar(text("SELECT 1"))
+
+ @api.app.get("/fixture-db-read")
+ async def read():
+ return await query(settings.AsyncSession)
+
+ fixture = request._fixturemanager.getfixturedefs("in_process_execution_api", request.node)[-1].func
Review Comment:
**[nit]** The tests of the test fixtures reach into pytest internals.
`request._fixturemanager.getfixturedefs(...)[-1].func` here, and `fresh_async_engine.__wrapped__()` in `test_secrets_resolver.py`, call fixture functions directly through private or undocumented attributes. pytest disallows calling fixtures directly ([deprecation note](https://docs.pytest.org/en/stable/deprecations.html#calling-fixtures-directly)), and `_fixturemanager` / `getfixturedefs` are not public API (the `getfixturedefs` signature already changed in pytest 8.1). A pytest or pytest-asyncio upgrade can therefore break these two tests while the product code is fine. Together they add about 130 lines that test scaffolding rather than the change.
The fixture's guarantees (a fresh engine per server loop, the previous pool left intact, globals restored) are already exercised by its consumers, `test_processor.py` and `test_triggerer_job.py`. If you want to keep explicit checks of the fixture, they can live under `tests_common` with the fixture extracted to a plain context manager that both the fixture and the tests call.
##########
airflow-core/src/airflow/secrets/metastore.py:
##########
@@ -23,15 +23,40 @@
from sqlalchemy import or_, select
+from airflow._shared.secrets_masker import mask_secret
from airflow.secrets import BaseSecretsBackend
-from airflow.utils.session import NEW_SESSION, provide_session
+from airflow.utils.session import NEW_SESSION, create_session_async, provide_session
if TYPE_CHECKING:
+ from sqlalchemy.engine import RowMapping
+ from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session
from airflow.models import Connection
+def _build_connection(values: RowMapping) -> Connection:
Review Comment:
**[nit]** `_build_connection` and `_get_variable_value` re-implement what the ORM load already does in the sync twins.
`get_connection` / `get_variable` above load the entity with `session.scalar(select(Connection)...)`, `session.expunge(...)` it, and let the `@reconstructor` (`on_db_load`) do the masking. The async versions select table columns and rebuild the object by hand: `Connection()` followed by a `setattr` per mapped column, then `on_db_load()`, and a `Variable` whose `_val` / `is_encrypted` are poked directly. That is a second loader per model that has to follow every future model change (a `column_property`, a deferred column, a new `__init__` side effect), while `AsyncSession` supports the same ORM select + expunge ([SQLAlchemy asyncio docs](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)):
```python
conn = await session.scalar(select(Connection).where(...).limit(1))
if conn:
session.expunge(conn)
return conn
```
Is there a reason the borrowed request session must not see the entity at all, beyond what `expunge` gives? If so, a one-line comment on the helpers saying why would help the next reader.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on PR #73414:
URL: https://github.com/apache/airflow/pull/73414#issuecomment-6065589427
@ashb could you please take another look?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4179028452
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
Review Comment:
First though: I'm really not sure about this name
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
Review Comment:
getboolean in a thread pool seems overkill?
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
+ raise ValueError(
+ "Multi-team mode is not configured in the Airflow environment but the task trying "
+ f"to access the {resource_name} belongs to a team"
+ )
+
+
+async def resolve_variable(
+ key: str, team_name: str | None = None, *, session: AsyncSession | None = None
+) -> str:
+ """Resolve a Variable without blocking the caller's event loop."""
+ await _validate_team_name(team_name, "variable")
+
+ SecretCache = _get_secret_cache()
+
+ try:
+ value = await run_in_threadpool(SecretCache.get_variable, key, team_name=team_name)
+ except SecretCache.NotPresentException:
+ pass
+ else:
+ if value is None:
+ raise KeyError(f"Variable {key} does not exist.")
+ await run_in_threadpool(mask_secret, value, key)
Review Comment:
Ditto here, adding a mask isn't really an expensive operation.
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
+ raise ValueError(
+ "Multi-team mode is not configured in the Airflow environment but the task trying "
+ f"to access the {resource_name} belongs to a team"
+ )
+
+
+async def resolve_variable(
+ key: str, team_name: str | None = None, *, session: AsyncSession | None = None
+) -> str:
+ """Resolve a Variable without blocking the caller's event loop."""
+ await _validate_team_name(team_name, "variable")
+
+ SecretCache = _get_secret_cache()
+
+ try:
+ value = await run_in_threadpool(SecretCache.get_variable, key, team_name=team_name)
Review Comment:
Isn't the cache in memory? If so why can't we call it directly?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #73414:
URL: https://github.com/apache/airflow/pull/73414#issuecomment-6273221338
LGTM, let's await @ashb review before approval.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4324335273
##########
airflow-core/docs/security/secrets/secrets-backend/index.rst:
##########
@@ -161,6 +161,72 @@ section of ``airflow.cfg``.
Additional arguments to your SecretsBackend can be configured in ``airflow.cfg`` by supplying a JSON string to ``backend_kwargs``, which will be passed to the ``__init__`` of your SecretsBackend.
See :ref:`Configuration <secrets_backend_configuration>` for more details, and :ref:`SSM Parameter Store <ssm_parameter_store_secrets>` for an example.
+Async server-side reads
+^^^^^^^^^^^^^^^^^^^^^^^
+
+When the Execution API serves a Variable or Connection, it resolves the value with
+``airflow.secrets.resolver.resolve_variable`` or ``airflow.secrets.resolver.resolve_connection``.
+Both functions search the configured backends in the same order as synchronous reads. There is no
+separate async configuration.
+
+For backends configured on the Execution API server, either or both of these optional methods
+enable native async reads:
+
+.. code-block:: python
+
+ async def aget_variable(self, key: str, team_name: str | None = None) -> str | None: ...
+
+
+ async def aget_connection(self, conn_id: str, team_name: str | None = None) -> Connection | None: ...
+
+They take the same arguments and return the same values as ``get_variable`` and ``get_connection``.
+``aget_connection`` returns the core :py:class:`airflow.models.connection.Connection` or ``None``.
+
+The Execution API server's resolver calls these methods as follows. These precedence and
+``team_name`` forwarding rules apply to server-side resolution:
+
+* **Precedence:** the resolver uses the async method when it is defined in the same class as its
+ synchronous counterpart, or in a subclass of that class. If a subclass overrides only the
+ synchronous method, the resolver uses that override. Existing custom backends therefore keep their
+ behavior when a parent class gains an async method.
+* **Fallback:** a backend without an async method needs no change. The resolver runs its synchronous
+ method in a worker thread.
+* **Errors:** the resolver doesn't retry a failing async method synchronously. It continues with the
+ next backend, as it does for synchronous methods. ``AirflowSecretsBackendAccessDenied`` ends the
+ search.
+* **Event loop:** async methods run on the event loop that serves the request. Create, use, and close
+ async clients on that loop. Don't create a loop-bound client in ``__init__``, and don't share it
+ with the synchronous method.
+* **Database session:** the metastore backend's async methods use the request's async database
+ session when the Execution API supplies one. Otherwise, they open and close their own. The resolver
+ never passes that session to another backend, and it doesn't commit or close it.
+* **Cancellation:** a cancelled request stops the backend search and the cache update. The resolver
+ can't interrupt a synchronous method that is already running in a worker thread. That method
Review Comment:
Python cannot forcibly stop a running thread. AnyIO supports [abandoning the wait](https://anyio.readthedocs.io/en/stable/api.html#anyio.to_thread.run_sync) and [cooperative cancellation](https://anyio.readthedocs.io/en/stable/threads.html#reacting-to-cancellation-in-worker-threads), but the latter requires checks inside the backend. The resolver uses Starlette's [`run_in_threadpool`](https://github.com/Kludex/starlette/blob/1.9.0/starlette/concurrency.py), which keeps AnyIO's default shielding.
Will rephrase it so as not to overstate when cancellation stops the search. It does not change the runtime policy though.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4183578508
##########
airflow-core/src/airflow/secrets/async_resolution.py:
##########
@@ -0,0 +1,215 @@
+# 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.
+"""Asynchronous server-side resolution through configured secrets backends."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+from starlette.concurrency import run_in_threadpool
+
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
+from airflow._shared.secrets_masker import mask_secret
+from airflow.configuration import conf, ensure_secrets_loaded
+from airflow.exceptions import AirflowNotFoundException
+from airflow.models.connection import AirflowSecretsBackendAccessDenied, Connection
+from airflow.secrets.metastore import MetastoreBackend
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+ from sqlalchemy.ext.asyncio import AsyncSession
+
+log = logging.getLogger(__name__)
+
+
+def _get_secret_cache() -> Any:
+ from airflow.sdk import SecretCache # noqa: SDK001
+
+ return SecretCache
+
+
+def _find_method_owner(backend: Any, method_name: str) -> int | None:
+ if method_name in vars(backend):
+ return -1
+ for index, cls in enumerate(type(backend).__mro__):
+ if method_name in vars(cls):
+ return index
+ return None
+
+
+def _get_native_async_method(backend: Any, *, async_name: str, sync_name: str) -> Callable[..., Any] | None:
+ if inspect.getattr_static(backend, async_name, None) is None:
+ return None
+ async_method = getattr(backend, async_name, None)
+ if async_method is None:
+ return None
+
+ async_owner = _find_method_owner(backend, async_name)
+ sync_owner = _find_method_owner(backend, sync_name)
+ if async_owner is not None and sync_owner is not None and sync_owner < async_owner:
+ return None
+ return async_method
+
+
+async def _call_backend_method(
+ method: Callable[..., Any],
+ *,
+ team_name: str | None,
+ session: AsyncSession | None = None,
+ **kwargs: Any,
+) -> Any:
+ if session is not None and accepts_kwarg(method, "session"):
+ kwargs["session"] = session
+
+ if inspect.iscoroutinefunction(method):
+ result = call_secrets_backend_method(method, team_name=team_name, **kwargs)
+ else:
+ result = await run_in_threadpool(
+ call_secrets_backend_method,
+ method,
+ team_name=team_name,
+ **kwargs,
+ )
+ if inspect.isawaitable(result):
+ return await result
+ return result
+
+
+async def _validate_team_name(team_name: str | None, resource_name: str) -> None:
+ if team_name and not await run_in_threadpool(conf.getboolean, "core", "multi_team"):
+ raise ValueError(
+ "Multi-team mode is not configured in the Airflow environment but the task trying "
+ f"to access the {resource_name} belongs to a team"
+ )
+
+
+async def resolve_variable(
+ key: str, team_name: str | None = None, *, session: AsyncSession | None = None
+) -> str:
+ """Resolve a Variable without blocking the caller's event loop."""
+ await _validate_team_name(team_name, "variable")
+
+ SecretCache = _get_secret_cache()
+
+ try:
+ value = await run_in_threadpool(SecretCache.get_variable, key, team_name=team_name)
Review Comment:
An initialized cache uses `multiprocessing.Manager().dict()`, so reads and writes communicate with the manager process. The resolver now calls an uninitialized cache directly, where calls only miss or do nothing, and uses a worker for the initialized cache.
`_call_secret_cache` checks `SecretCache._cache` because there is no public initialization predicate, and the SDK cache API is unchanged.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on PR #73414:
URL: https://github.com/apache/airflow/pull/73414#issuecomment-5993732981
You're right: main already has the API lifespan cleanup from #73838 and an Execution API `async_db_engine` fixture. I've corrected the stale description and reused that cleanup.
The remaining case is the in-process API used by the Dag processor and triggerer tests. Async reads exposed pooled connections opened on another test's loop. The focused `in_process_execution_api` fixture gives these tests a fresh engine and uses `TestClient` to own the private app's lifespan and loop. Its synchronous transport uses that loop; the captured engine is disposed before the loop stops, and the previous globals are restored even if setup or cleanup fails: [in-process fixture](https://github.com/apache/airflow/blob/d69564c1a62fa51ad0d3a3450ede7aaed9c9750f/airflow-core/tests/conftest.py).
Regression tests cover delayed transport collection, a populated previous pool remaining intact and usable, startup and shutdown failures, disposal failure, and partial setup. The direct async fixture also restores its globals on failure. Locally, 136 focused tests passed on each of SQLite, PostgreSQL and MySQL. Another 130 GET contract cases passed per backend across all ten supported API versions. Remote CI for revised head `d69564c1a6` is pending.
I dropped the earlier separate production-ownership design after finding no production counterexample to the existing loop ownership. This PR targets main directly and keeps the existing production session and engine infrastructure.
---
Drafted-by: Codex (GPT-6) (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #73414:
URL: https://github.com/apache/airflow/pull/73414#issuecomment-5984112661
> Disposes the test async engine inside each TestClient event loop, preventing pooled MySQL connections from crossing loop lifetimes.
I think we already have something similar for this in the Exec API tests. Maybe it's time to make it apply to all FastAPI test
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4308099081
##########
airflow-core/tests/unit/api_fastapi/execution_api/test_app.py:
##########
@@ -96,6 +97,96 @@ def test_server_and_in_process_apps_isolate_auth(in_process_first, monkeypatch):
assert authenticated_response.json() == response.json()
+@pytest.mark.parametrize("failure", [None, "startup", "shutdown", "dispose", "configure"])
+@mock.patch("airflow.api_fastapi.execution_api.app.InProcessExecutionAPI", autospec=True)
+def test_in_process_fixture_owns_engine_lifetime(mock_api, request, failure):
+ previous_engine, previous_factory = settings.async_engine, settings.AsyncSession
+ api = InProcessExecutionAPI()
+ mock_api.return_value = api
+ original_lifespan = api.app.router.lifespan_context
+ disposed = []
+ original_dispose = AsyncEngine.dispose
+ original_configure = settings._configure_async_session
+
+ @asynccontextmanager
+ async def lifespan(app):
+ async with original_lifespan(app):
+ if failure == "startup":
+ raise RuntimeError("startup failed")
+ yield
+ if failure == "shutdown":
+ raise RuntimeError("shutdown failed")
+
+ api.app.router.lifespan_context = lifespan
+
+ async def dispose(engine, *args, **kwargs):
+ disposed.append((engine, asyncio.get_running_loop()))
+ await original_dispose(engine, *args, **kwargs)
+ if engine is not previous_engine and failure == "dispose":
+ raise RuntimeError("dispose failed")
+
+ def configure():
+ original_configure()
+ if failure == "configure":
+ raise RuntimeError("configure failed")
+
+ async def query(factory):
+ async with factory() as session:
+ return await session.scalar(text("SELECT 1"))
+
+ @api.app.get("/fixture-db-read")
+ async def read():
+ return await query(settings.AsyncSession)
+
+ fixture = request._fixturemanager.getfixturedefs("in_process_execution_api", request.node)[-1].func
Review Comment:
Removed both fixture-introspection tests, including request._fixturemanager/getfixturedefs and fresh_async_engine.__wrapped__. The fixtures are still consumed through normal pytest injection, and the existing Dag processor and triggerer integration tests retain coverage of their use.
---
Drafted-by: Codex (GPT-6) (reviewed by @Dev-iL)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make Execution API secret reads async [airflow]
Posted by "Dev-iL (via GitHub)" <gi...@apache.org>.
Dev-iL commented on code in PR #73414:
URL: https://github.com/apache/airflow/pull/73414#discussion_r4308098734
##########
airflow-core/docs/security/secrets/secrets-backend/index.rst:
##########
@@ -161,6 +161,62 @@ section of ``airflow.cfg``.
Additional arguments to your SecretsBackend can be configured in ``airflow.cfg`` by supplying a JSON string to ``backend_kwargs``, which will be passed to the ``__init__`` of your SecretsBackend.
See :ref:`Configuration <secrets_backend_configuration>` for more details, and :ref:`SSM Parameter Store <ssm_parameter_store_secrets>` for an example.
+Async server-side reads
+^^^^^^^^^^^^^^^^^^^^^^^
+
+When the Execution API serves a Variable or Connection, it resolves the value with
+``airflow.secrets.resolver.resolve_variable`` or ``airflow.secrets.resolver.resolve_connection``.
+Both functions search the configured backends in the same order as synchronous reads. There is no
+separate async configuration.
+
+A backend can implement either or both of these optional methods:
+
+.. code-block:: python
+
+ async def aget_variable(self, key: str, team_name: str | None = None) -> str | None: ...
+
+
+ async def aget_connection(self, conn_id: str, team_name: str | None = None) -> Connection | None: ...
+
+They take the same arguments and return the same values as ``get_variable`` and ``get_connection``.
+``aget_connection`` returns the core :py:class:`airflow.models.connection.Connection` or ``None``.
+
+The resolver calls these methods as follows:
+
+* **Precedence:** the resolver uses the async method when it is defined in the same class as its
Review Comment:
Scoped the contract explicitly to backends configured on the Execution API server. The page now also describes local worker lookups: environment/worker backends precede ExecutionAPISecretsBackend, the SDK prefers an available async method even over a subclass's sync override, and those local calls do not receive team_name. The SDK behavior is unchanged.
---
Drafted-by: Codex (GPT-6) (reviewed by @Dev-iL)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org