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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
