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()
[email protected]("failure", [None, "startup", "shutdown", "dispose",
"configure"])
[email protected]("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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
