[dabla]

Add async Variable methods (aget, aset, adelete, akeys) with tests

This PR adds async counterparts to all synchronous Variable methods.

What changed

Variable class (task-sdk/src/airflow/sdk/definitions/variable.py)

Four new async class methods:

Async method	Sync equivalent	Notes
Variable.aget	Variable.get	Awaits _async_get_variable; falls back to default (masked) on VARIABLE_NOT_FOUND
Variable.aset	Variable.set	Awaits _async_set_variable
Variable.adelete	Variable.delete	Awaits _async_delete_variable
Variable.akeys	Variable.keys	Awaits _async_get_variable_keys and returns the resolved list; unlike keys it is not lazy

Context helpers (task-sdk/src/airflow/sdk/execution_time/context.py)

This PR also adds the five helpers behind those methods. Each mirrors its sync twin:

• _async_mask_and_deserialize_variable masks the raw value and, with
  deserialize_json, the decoded value through amask_secret.
• _async_get_variable checks SecretCache, then asks each secrets backend:
  aget_variable when the backend has one, otherwise get_variable via
  asyncio.to_thread. ExecutionAPISecretsBackend.aget_variable reaches the
  Execution API via asend. An authoritative deny
  (AirflowSecretsBackendAccessDenied) is re-raised and never falls through to
  the next backend.
• _async_set_variable warns when a worker-side secrets backend already holds
  the key (checked via asyncio.to_thread), sends PutVariable via asend,
  and invalidates the cache.
• _async_delete_variable sends DeleteVariable via asend and invalidates
  the cache.
• _async_get_variable_keys pages through GetVariableKeys requests via asend.

Tests

test_variables.py

TestAsyncVariables and TestAsyncVariableKeys mirror the sync test cases:

• test_avar_get: plain and JSON-deserialized values; verifies the GetVariable request
• test_avar_get_returns_default_when_not_found, test_avar_get_raises_when_not_found_without_default
• test_avar_set: plain and JSON-serialized values; verifies the PutVariable payload
• test_avar_delete: verifies DeleteVariable is sent via asend
• test_akeys: prefix filtering (all / with prefix / empty)
• test_akeys_paginates_when_results_exceed_page_size
• test_akeys_raises_on_error_response
• test_akeys_raises_on_unexpected_response_type

test_context.py

TestAsyncVariableContext tests the context helpers directly:

• _async_get_variable: from the API, from a secrets backend, not found raises,
  a PERMISSION_DENIED from the Execution API raises and the next backend is never asked
• masking: cache hit, plain value, JSON dict / list / string values, sensitive key, invalid JSON
• _async_set_variable: plain value, JSON-serialized value, warning on a conflicting backend
• _async_delete_variable: sends DeleteVariable
• _async_get_variable_keys: prefix filtering, pagination, error, unexpected response

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  GitHub Copilot (Claude Sonnet 4.6), Claude Code

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[amoghrajesh]

Requesting changes to avoid accidental merge, directionally good, but there's work needed.

[amoghrajesh]

Looking better, some comments.

[amoghrajesh]

The PR body is wrong. It mentions "The four underlying async* functions were already implemented" and lists them as background. This PR adds all four, along with _async_mask_and_deserialize_variable, please update it.

Looks good otherwise with 2 nits.