Fix PydanticAIHook.get_hook losing the caller's conn_id with secrets-backend aliases

[dabla]

Fix LLMOperator / @task.agent failing with secrets-backend connection aliases

Context

We use a custom secrets backend provider built on top of HashiCorp Vault that
introduces environment-independent connection aliases.

Dag authors never hardcode environment-specific connection ids
(e.g. openai-dev, openai-prod). Instead they reference a single logical
alias (e.g. my-llm) that is stable across every environment. The secrets
backend knows at runtime which environment it is deployed in and resolves the
alias to the correct underlying Vault secret — returning a Connection whose
conn_id is the canonical, environment-specific id (openai-prod,
openai-dev, …).

This keeps Dags 100% environment-agnostic: the same conn_id="my-llm" works
in development, staging, and production without any per-environment
configuration in the Dag code.

Problem

When get_connection("my-llm") is called, the secrets backend returns a
Connection object whose conn_id has been rewritten to the resolved,
canonical id (e.g. "openai-prod").

BaseHook.get_hook passed that Connection straight to connection.get_hook(),
which instantiated the hook with llm_conn_id = "openai-prod". Any subsequent
get_conn() call (e.g. inside LLMOperator / @task.agent) then tried to
fetch "openai-prod" from the secrets backend — a key it does not recognise,
because only the alias "my-llm" is registered — and raised
AirflowNotFoundException.

Fix

Override get_hook in PydanticAIHook to call super().get_hook() and then
immediately restore the caller's original conn_id (the alias) on the hook:

@classmethod
def get_hook(cls, conn_id: str, hook_params: dict | None = None) -> PydanticAIHook:
    hook = super().get_hook(conn_id, hook_params=hook_params)
    setattr(hook, hook.conn_name_attr, conn_id)
    return hook

Why override here and not only in BaseHook?

BaseHook applies the same fix starting from Airflow 3.4. However,
providers are independently releasable — a user on Airflow 3.2 or 3.3 can
upgrade this provider without upgrading Airflow core. Keeping the override in
PydanticAIHook ensures the fix is available on all supported Airflow
versions as soon as the provider is updated.

The two fixes are intentionally complementary:

Airflow version	BaseHook rekeyed?	PydanticAIHook override rekeyed?	Net result
< 3.4	✗	✓	fixed via provider override
≥ 3.4	✓	✓ (no-op setattr)	fixed via BaseHook

Impact

• Unblocks LLMOperator and @task.agent when connections are stored under
  environment-independent aliases in a custom secrets backend, on any supported
  Airflow version.
• No behaviour change when the secrets backend does not rewrite conn_id
  (the setattr is a no-op in that case).
• No changes to the secrets-backend infrastructure.

---

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

• [ x ] Yes (please specify the tool below)

Claude Sonnet 4.6

---

• 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]

Before making the BaseHookchange, it is worth checking whether this is actually needed. Airflow's secrets-backend contract already guarantees that Connection.conn_id on the object returned by get_connection() matches the requested lookup key, see: https://github.com/apache/airflow/blob/main/shared/secrets_backend/src/airflow_shared/secrets_backend/base.py#L121-L157

Both construct the Connection using the requested conn_id, not anything derived from the secret's contents. Every stock backend such as EnvironmentVariablesBackend, MetastoreBackend, LocalFilesystemBackend, and the Hashicorp VaultBackend only override get_conn_valueand never touches conn_id. That is how alias style translation is supposed to work: the backend uses the alias to find the right underlying secret, but the Connection object it hands back still identifies itself by the alias.

The custom backend that you describe in this PR might be overriding get_connection/deserialize_connection (or constructing Connection some other way) and explicitly stamping the canonical, environment specific id onto conn_id which is a deviation from the contract every other backend honors, not a gap in BaseHook as I see it

Can you check in the custom backend for above and have it return Connection(conn_id=<alias>, ...) and keep the canonical/resolved id purely as an internal lookup detail?

[dabla]

Before making the BaseHookchange, it is worth checking whether this is actually needed. Airflow's secrets-backend contract already guarantees that Connection.conn_id on the object returned by get_connection() matches the requested lookup key, see: https://github.com/apache/airflow/blob/main/shared/secrets_backend/src/airflow_shared/secrets_backend/base.py#L121-L157

Both construct the Connection using the requested conn_id, not anything derived from the secret's contents. Every stock backend such as EnvironmentVariablesBackend, MetastoreBackend, LocalFilesystemBackend, and the Hashicorp VaultBackend only override get_conn_valueand never touches conn_id. That is how alias style translation is supposed to work: the backend uses the alias to find the right underlying secret, but the Connection object it hands back still identifies itself by the alias.

The custom backend that you describe in this PR might be overriding get_connection/deserialize_connection (or constructing Connection some other way) and explicitly stamping the canonical, environment specific id onto conn_id which is a deviation from the contract every other backend honors, not a gap in BaseHook as I see it

Can you check in the custom backend for above and have it return Connection(conn_id=<alias>, ...) and keep the canonical/resolved id purely as an internal lookup detail?

Good point @amoghrajesh, I will try this change in our custom provider, if it fixes the issue, then I'll close this PR.