Build deferred AWS hooks from the operator's own settings

[SEPURI-SAI-KRISHNA]

Addresses the bulk of the deferred hook-configuration gap tracked in #72144.

AwsBaseWaiterTrigger now builds the hook itself, from the parameters it already serializes, driven by an aws_hook_class attribute, the same arrangement AwsBaseHookMixin gives the operators. Subclasses name the hook they need instead of constructing it, so region_name, verify and botocore_config reach the triggerer by default rather than only where a subclass remembered to thread them through.

That removes 40 bespoke hook() implementations, and takes the provider from 49 of 113 defer sites forwarding the full hook configuration to 110 of 113.

A subclass that declares neither aws_hook_class nor its own hook() is now rejected in __init_subclass__, so it fails when the class is created rather than later in the triggerer, which is the only place hook() is reached from.

Merge order

#72098 → this → #72449 / #72472 (those last two in either order).

• Use the operator's AWS settings for deferred Neptune, MWAA, SSM tasks #72098 is the only hard dependency, and it is containment rather than conflict. This PR carries a rebased copy of its two commits; all 8 of its files are a strict subset of the 33 here. It can merge whenever it is ready, and this PR gets 8 files smaller on the next sync.
• No file-level overlap with either workshop PR. This PR does not touch sensors/batch.py, sensors/opensearch_serverless.py or triggers/opensearch_serverless.py, and neither of those PRs touches the invariant test. There is no merge conflict in any ordering.
• The only coupling is the allowlist: whichever of Pass verify and botocore_config to BatchJobTrigger in BatchSensor (#72278) #72449 / Preserve OpenSearch Serverless hook configuration when deferring #72472 merges second drops its own line from PENDING_MIGRATION (and, for Preserve OpenSearch Serverless hook configuration when deferring #72472, from HAND_BUILT_HOOK_EXCEPTIONS). The assertion prints the exact line to remove.

What is deliberately left out

Two services are reserved for the Contributors Workshop, at the request of the workshop organiser on #72144: sensors/batch.py and sensors/opensearch_serverless.py. They are listed in PENDING_MIGRATION in the invariant test, which asserts each entry is still needed, a stale line fails the suite, so the allowlist cannot outlive the work it tracks. They are covered by #72449 and #72472 respectively.

Two defer sites are not instances of this bug at all, and are named in UNCONFIGURABLE_TRIGGERS rather than passed over silently. SageMakerNotebookOperator defers to SageMakerNotebookJobTrigger, a plain BaseTrigger whose hook is addressed by execution name and takes no connection parameters. EksPodOperator defers to EksPodTrigger, a KubernetesPodTrigger that reaches the pod through a kubeconfig rather than a boto3 client.

Invariant test

test_deferred_hook_configuration.py walks every self.defer(trigger=...) call in the provider and fails if one does not pass the hook configuration, plus asserts every AwsBaseWaiterTrigger subclass can actually build a hook. It resolves a trigger= expression to every construction it can evaluate to, so a trigger chosen in a conditional expression is checked on both branches, that is how the two EmrContainer sites were caught. A defer site whose trigger is a bare reference is asserted against an explicit allowlist rather than skipped. An operator added later that forgets the parameters fails in CI rather than in production.

Moving the four purely static checks into a prek hook is a good idea and is planned as a follow-up; the concrete correctness fixes from review (POSIX-normalised paths, ast.Attribute callees, MRO-aware hook() detection) are already in here.

Notes for review

• aws_hook_class binds the hook at class definition, so the @patch("...triggers.<module>.<Hook>") idiom no longer intercepts it for migrated triggers. No existing test needed changing as a result.
• EmrContainerTrigger's hook takes an extra virtual_cluster_id, so it overrides _hook_parameters rather than using the default, the escape hatch the base class keeps for exactly this.
• EksDeleteClusterTrigger bypasses the base __init__ and rolls its own serialize(), so it sets and serializes the two new parameters explicitly. That it skips super().__init__() at all is pre-existing and worth a follow-up.
• A changelog warning is included: the triggerer now uses the operator's region, SSL verification and botocore config instead of boto3 defaults, which is observable for deployments that relied on the triggerer host's own defaults.
• Verified against a full run of the operator, sensor and trigger suites at this commit: 2528 passed, 5 skipped, 0 failures, 29 collection errors. Every one of the 29 is ModuleNotFoundError: No module named 'airflow_shared', an optional dependency missing from my local environment, in files this PR does not touch.

---

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

• Yes, Claude Code (Opus 5)

Generated-by: Claude Code (Opus 5) following the guidelines

[potiuk]

cc: @ferruzzi @vincbeck @o-nikolas

[vincbeck]

@seanghaeli @ramitkataria

[o-nikolas]

Nice work — the design is right and the migration is genuinely mechanical. Two things I'd like addressed before merge, plus a merge-order hazard that isn't your fault.

aws_hook_class + _hook_parameters + a base hook() is a deliberate mirror of AwsBaseHookMixin, down to naming the botocore config config in the parameter mapping, so anyone who has worked on the operator side will recognise it immediately. EmrContainerTrigger overriding _hook_parameters to add virtual_cluster_id reads better than the hook() it replaces.

I spot-checked the removed hook() bodies for behaviour changes and found none: EmrServerlessHook(self.aws_conn_id) passed positionally still lands on aws_conn_id, and GlueJobHook is constructed without job_name both before and after, so the thick-hook defaults are unchanged. The DMS triggers are a good catch — they did pass verify/config in hook(), but their __init__ never accepted them, so those were structurally always None.

1. The failure mode moved into the triggerer

hook() now raises AttributeError when aws_hook_class is unset, but hook() is only called from run(), which runs in the triggerer. @abstractmethod used to fail at instantiation, i.e. in the worker at defer time. Now a subclass that forgets both aws_hook_class and hook() defers successfully and fails later, somewhere less visible.

The operator side doesn't have this problem: AwsBaseOperator.__init__ and AwsBaseSensor.__init__ both call validate_attributes() (operators/base_aws.py:101, sensors/base_aws.py:101). I'd add the equivalent here, ideally in __init_subclass__ so it fails at import time and doesn't depend on __init__ being reached — which matters because EksDeleteClusterTrigger bypasses super().__init__() entirely.

Related: the check is hasattr only, while the mixin's validate_attributes also asserts issubclass(..., AwsGenericHook). Worth matching — a typo'd assignment currently fails with a confusing TypeError from calling a non-class.

2. The static half of the invariant test belongs in a prek hook

The invariant is the reason a 40-file mechanical change is safe to review, so I'm glad it's here. But four of the five tests are pure AST analysis of source files with no runtime component, and there's already a home for exactly that: scripts/ci/prek/check_trigger_serialize_init.py does the same class of static check on triggers, alongside a dozen or so other AST-walking hooks in that directory. As a prek hook it runs on every commit, fails with a file:line message rather than a parametrized test id, and doesn't need the provider's optional dependencies installed.

I'd move test_defer_sites_are_discovered, test_no_defer_site_escapes_the_check, test_deferred_trigger_receives_hook_configuration and test_hand_built_trigger_hook_receives_configuration there, keeping only test_waiter_trigger_can_build_a_hook as a unit test since that one genuinely needs imports.

Concrete fragilities, wherever it ends up living:

• find_waiter_triggers() imports every trigger module at collection time. triggers/eks.py imports KubernetesPodTrigger from cncf.kubernetes; if that isn't installed the ImportError is a collection error for the whole file rather than a skip. The "30 collection errors locally" in your description is this failure mode.
• find_defer_sites and find_unreadable_defer_sites duplicate ~20 lines of identical self.defer matching. One walk returning both categories.
• str(path.relative_to(AWS_ROOT)) yields backslashes on Windows, so every ("sensors/batch.py", ...) comparison silently stops matching. .as_posix() fixes it. Not a CI concern, but contributors do run these locally.
• find_hand_built_hooks matches only ast.Name callees ending in Hook, while find_defer_sites handles ast.Attribute too, so a module.SomeHook(...) construction escapes the check.
• "hook" in vars(trigger_class) only inspects the class's own __dict__, so a subclass inheriting an overridden hook() from an intermediate base would fail spuriously.
• HAND_BUILT_HOOK_EXCEPTIONS keys on bare filename rather than relative path. Fine today, collides the day two trigger modules share a name.

3. Merge-order hazard across three PRs

PENDING_MIGRATION asserts each entry is still needed. #72472 (the OpenSearch Serverless workshop task) is green and close to ready, and it drops the hook configuration into that exact defer site — so if it merges first, this PR's suite fails until the entry is removed. Combined with the #72098 dependency the intended order is #72098 → this → #72472, and it would help to say so explicitly in the description, since it currently only mentions #72098.

The self-detecting design does work in the other direction: assert missing in the allowlist branch means a stale entry fails loudly rather than rotting quietly. Good call.

4. Description doesn't match the code

The description says three services are reserved for the workshop and "are listed in PENDING_MIGRATION", but only two are — sensors/batch.py and sensors/opensearch_serverless.py. operators/sagemaker_unified_studio_notebook.py is instead in UNCONFIGURABLE_TRIGGERS with the rationale "takes no connection parameters at all", which reads as contradicting the paragraph above it. Worth reconciling, since #72144 points contributors at this list.

5. Changelog

This changes runtime behaviour at roughly 60 additional defer sites: the triggerer half now uses the operator's region, SSL verification and botocore config instead of boto3 defaults. That's the fix everyone wants, but it is observable — a deployment whose triggerer happened to work because it fell back to its own default region will now be pointed elsewhere. Providers don't consume newsfragments, so this warrants a line in providers/amazon/docs/changelog.rst rather than being left for the release manager to infer from git log.

Minor

• Docstring wording for the three new params drifts per file (emr.py: "The AWS region where the resources to watch are"; eks.py: "Which AWS region the connection should use"). Pre-existing variance, not worth churning.
• EksCreateClusterTrigger puts region_name in serialized_fields and gets it from the base's prune_dict block. Harmless duplicate and pre-existing, but a clean follow-up now that the base owns more of this.
• EksDeleteClusterTrigger still never calls super().__init__(), so BaseTrigger.__init__ doesn't run, and it now hand-rolls the base's prune_dict serialization too. Pre-existing, but this PR is the natural moment to note it as a follow-up.

I didn't run the suite locally; relying on the green CI here, which is the meaningful signal given your local environment was missing optional dependencies.

---

Drafted-by: Kiro CLI (Opus 5); reviewed by @o-nikolas before posting

[o-nikolas]

The changes are fairly mechanical and are looking good. An automated review with our magpie skill found a few things that I think you should address (see the comment above).

I'm also still trying to wrap my head around all related PRs and what their merge order should be. I think this is becoming more complicated than it needs to be. Between the three (#72098, #72171, #72472) what should the order be?

[SEPURI-SAI-KRISHNA]

Thanks, this is a genuinely useful review. Addressed 1, 4 and 5 in the commit above; concrete answers to 2 and 3 below.

1. Failure mode moved into the triggerer: fixed

You are right, and EksDeleteClusterTrigger bypassing super().__init__() is exactly why I put the check in __init_subclass__ rather than __init__. It now fails when the class is created, so an unusable trigger cannot be imported, let alone deferred to.

The check accepts a subclass that declares aws_hook_class or overrides hook(), and it tests the override as cls.hook is not AwsBaseWaiterTrigger.hook, so an override inherited from an intermediate base still counts. It also asserts issubclass(..., AwsGenericHook) as you asked, so a typo'd assignment now says so instead of raising TypeError from calling a non-class. That check needs AwsGenericHook at runtime, so its import moves out of the TYPE_CHECKING block. It is the same module-level import utils/mixins.py already carries, and nothing in hooks/base_aws.py imports triggers, so there is no cycle. Four tests in triggers/test_base.py cover both rejections and both accepted shapes. Every existing trigger imports clean.

2. Moving the static half to a prek hook: agreed, but as a follow-up

I have fixed the three that are real bugs regardless of where the code lives:

• .as_posix() instead of str(path.relative_to(...)), so the allowlist tuples keep matching on Windows.
• find_hand_built_hooks now matches an ast.Attribute callee, so a module.SomeHook(...) construction cannot slip past. It finds no new sites today; it closes the hole.
• The "hook" in vars(...) check is now the MRO-aware comparison described above.

On the move itself, I would rather do it in a follow-up than here, for two reasons. This PR is already 33 files, and the invariant is the thing that makes the rest of them reviewable, so changing where it lives in the same diff means the evidence and the change move together. And a prek hook needs a new script plus a .pre-commit-config.yaml entry, which is a different review from a provider migration. I will open it once this lands.

The duplicated self.defer matching and the bare-filename keys in HAND_BUILT_HOOK_EXCEPTIONS are both worth doing and I will fold them into that follow-up.

On the collection errors: I do not think find_waiter_triggers() is the cause. That file collects and runs clean here, 244 passed and 5 skipped. At this commit the full operator, sensor and trigger run is 2528 passed, 5 skipped, 0 failures, 29 collection errors, and every one of the 29 is ModuleNotFoundError: No module named 'airflow_shared' in files this PR does not touch. I have corrected the stale counts in the description, which is where the "30" you saw came from.

Worth flagging that moving the four AST tests to prek would not remove the import requirement either, since test_waiter_trigger_can_build_a_hook, the one you suggested keeping as a unit test, is precisely the one that imports every trigger module. If we want that robust for partial installs it needs its own guard, which I will include in the follow-up.

3. Merge order

#72098 -> this -> #72449 / #72472, those last two in either order. It is less entangled than it looks, and I should have said so in the description.

#72098 to this is the only hard dependency, and it is containment rather than conflict. This PR carries a rebased copy of #72098's two commits, and all 8 of its files are a strict subset of the 33 here. So #72098 can merge whenever it is ready on its own merits, and when it does, this PR simply gets 8 files smaller on the next sync. Nothing to coordinate.

This PR has no file-level overlap with either workshop PR. It does not touch sensors/batch.py, sensors/opensearch_serverless.py or triggers/opensearch_serverless.py, and neither workshop PR touches the test file here. There is no merge conflict in any ordering.

The whole coupling is three lines in one allowlist:

PR	line to delete when it merges second
#72449	("sensors/batch.py", "BatchJobTrigger") from PENDING_MIGRATION
#72472	("sensors/opensearch_serverless.py", "OpenSearchServerlessCollectionActiveTrigger") from PENDING_MIGRATION and ("opensearch_serverless.py", "OpenSearchServerlessHook") from HAND_BUILT_HOOK_EXCEPTIONS

It is symmetric, and the assertion prints the exact line to remove, so it is a one-line fix with no investigation. To keep it off your plate: I will watch all three and push the deletion myself, wherever it needs to land. Merge them in whatever order suits.

4. Description: fixed

You are right that it contradicted itself. Only two services are in PENDING_MIGRATION, and the paragraph immediately below explains why SageMaker is not an instance of this bug at all. Rewritten so the workshop list is the two that are actually allowlisted, with SageMaker and Eks described only as the not-applicable cases they are. Merge order added.

5. Changelog: added

Added a .. warning:: at the top of providers/amazon/docs/changelog.rst, alongside the existing ones, since that block is the hand-written part the release manager does not regenerate. It names what changes observably: the triggerer now uses the operator's region instead of the triggerer host's AWS_DEFAULT_REGION, and applies the SSL verification and botocore config that never reached it before.

Minor

Agreed on all three, and none touched here. EksCreateClusterTrigger's duplicate region_name and EksDeleteClusterTrigger skipping super().__init__() are both pre-existing and both a good fit for the follow-up, now that the base owns more of this. Leaving the docstring wording variance alone as you suggested.

---

Drafted-by: Claude Code (Opus 5); reviewed by @SEPURI-SAI-KRISHNA before posting

[SEPURI-SAI-KRISHNA]

Expanding here on the inline thread about trigger= expressions the sweep cannot read, since that thread is now outdated and collapses by default.

I checked the two cases reported there and then went through the guard for the other cases in the same category instead of fixing only the reported ones.

trigger= expression	constructions	flagged unreadable	outcome before
SomeTrigger(...)	1	no	covered
trigger (bare name)	0	yes	covered through the allowlist
self._trigger	0	no	silently missed
triggers[kind]	0	no	silently missed
A() if flag else self._t	1	no	worse: looked covered, but only one branch was checked
TRIGGERS[kind](x=1)	1	no	recorded with an empty class name

The fifth row was the main one I wanted to fix. It ends up in DEFER_SITES, so the site appears in the parametrized test and looks covered, but the unreadable branch contributes nothing. Someone looking at the test ids would see the file listed and assume it had been checked.

I found the sixth case while doing the sweep. A construction whose callee cannot be resolved is still a Call, so it was kept. However, PENDING_MIGRATION and UNCONFIGURABLE_TRIGGERS are keyed by class name. An unnamed construction can therefore never match either allowlist, and the failure would end up saying defers to without passing ....

I fixed this as you suggested. trigger_constructions now returns None instead of [] whenever the expression cannot be resolved. That covers bare references, attributes, subscripts, conditionals where either branch is unreadable, and calls where the callee cannot be named.

find_unreadable_defer_sites now checks for exactly that None result instead of checking for ast.Name. Anything the sweep cannot resolve therefore has to be added to UNREADABLE_DEFER_SITES or the suite fails. The entries use ast.unparse, so a future entry shows the actual expression.

I also cleaned up two things while I was there.

The two almost identical walks are now a single walk_defer_sites() generator that both functions use. This also removes the duplicated self.defer matching that came up earlier in the review.

The sweep also no longer skips files whose path.parent.name is not operators or sensors. defer is a BaseOperator method, so a site can appear anywhere. A nested subpackage would previously have been silently skipped. There are no such sites today, so the counts are unchanged: 113 defer sites, 44 hand built hook constructions, and one allowlisted operators/eks.py entry.

test_unreadable_trigger_expressions_resolve_to_none is parametrized over all eight shapes. Five of the eight fail with the previous implementation, so the test is checking the actual gap rather than just exercising the new code.

There are no affected provider sites today. Apart from the allowlisted bare name in operators/eks.py, the only trigger expressions that are not a plain SomeTrigger(...) call are the two EmrContainerTrigger conditionals in operators/emr.py and sensors/emr.py, where both branches are readable, and one self.trigger_class(...) in sensors/bedrock.py. So this was a latent hole in the guard rather than a current gap.

I also checked four related cases and left them unchanged:

• find_hand_built_hooks still relies on the callee name ending in Hook, so a hook created through a variable or self.hook_class would not be found. That is a limitation of the current heuristic and fixing it would need a different approach.

• find_waiter_triggers() imports every trigger module, so a partial install can turn an ImportError into a collection error. This was already noted earlier in the review and will go into the prek hook follow up, together with changing the exception lists to use relative paths instead of bare filenames. There are no colliding basenames today.

• A hook construction using **kwargs would report all three parameters as missing. That fails loudly rather than passing silently, so it is a false positive rather than a hole. There are no such cases today.

• read_trigger_name returns the attribute name for a callee like self.trigger_class(...) in sensors/bedrock.py, so that site is recorded as trigger_class rather than a real class name. It passes today because all three parameters are passed there, and a site like that would fail loudly rather than silently, so I left it as is.

Finally, defer() takes trigger as a keyword only argument, so the keyword lookup cannot miss a positional argument. That case is safe by construction.

---

Drafted-by: Claude Opus 5; reviewed by @SEPURI-SAI-KRISHNA before posting

[o-nikolas]

Looking good to me. Can you get the tests green and we can be ready to merge?

[SEPURI-SAI-KRISHNA]

Tests are green now. The only failing job was MyPy providers: the sweep resolved a construction's class name separately from deciding the construction was readable, so the name stayed optional at every use. Pairing each construction with the name that made it readable removes the optionality. 83 checks, no failures, and the branch is mergeable.

[potiuk]

Approving. The design is the right one — mirroring AwsBaseHookMixin means anyone who has worked on the operator side reads this without a second thought, and pushing the rejection into __init_subclass__ rather than __init__ is the correct answer to EksDeleteClusterTrigger never calling super().__init__().

I verified the three claims this change actually rests on rather than taking the description's word for them:

• No import cycle from moving AwsGenericHook out of TYPE_CHECKING: hooks/base_aws.py imports nothing from triggers, so the new runtime import in triggers/base.py is safe.
• aws_hook_class(**self._hook_parameters) really does construct for every hook named here. I checked all 13 __init__ signatures; each takes **kwargs, so aws_conn_id / region_name / verify / config land safely. EmrContainerHook is the one that needs more, and the _hook_parameters override covers it.
• The changelog entry is the right mechanism — AGENTS.md sends provider-visible notes to changelog.rst rather than a newsfragment, and a deferred task silently changing region is exactly the kind of thing that belongs there.

Two things for a follow-up rather than this PR; neither is worth another round trip on a change this size.

The invariant suite has one assertion that cannot fail

test_waiter_trigger_can_build_a_hook asserts a subclass declares aws_hook_class or overrides hook() — but find_waiter_triggers() imports every trigger module first, and __init_subclass__ already raises AttributeError at class creation for precisely that condition. A class that would fail the assertion cannot reach the parametrize list; collection dies first.

That matters only because of what it is named and what the description promises it does ("asserts every AwsBaseWaiterTrigger subclass can actually build a hook"). The risk it looks like it covers — a future thick hook whose __init__ does not accept the four forwarded keys — is unchecked. Nothing is broken today, as above. Detail inline.

Changelog placement after the rebase

Flagging rather than asserting, since you will know the release state better than I can infer it: the new warning sits above the first version heading, which was right against this branch's base, but main has since cut 9.36.0 and moved the Comprehend warning under it. Worth a glance after merge that the warning ended up where you intended.

Thanks for the invariant test and the merge-order write-up — 33 files is a lot to ask a reviewer to trust, and both are why this was reviewable at all.

---

This review was drafted by an AI-assisted tool and
confirmed by an Apache Airflow maintainer. The maintainer
approving this PR has read the findings and signed off. If
something feels off, please reply on the PR and a maintainer
will follow up.

More on how Apache Airflow handles maintainer review:
contributing-docs/05_pull_requests.rst.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[potiuk]

Two fixups for the follow-ups in my review above — one applicable inline, one that needs a rebase instead.

The changelog point is no longer the "worth a glance" I called it. I computed the three-way merge of changelog.rst (merge-base bdf2abc, this head, current main) and it resolves cleanly, but it lands the new warning inside the 9.36.0 section. 9.36.0 was written by 209e34c5 ("Prepare providers release 2026-09-09"), so it is a cut release, and this change is not in it. Merging as-is publishes the warning against a version that does not have the behaviour it describes. Detail inline; it needs a rebase, so there is no suggestion button for it.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting