Replace _handle_request elif chains with a handler registry #65827

[JH-A-Kim]

---

Description

This PR implements the changes specified in #65827

The goal of this pull request is to refactor the _handle_request method by replacing a long, isinstance elif chain with a Handler Registry design pattern. This improves maintainability, and prevents future code bloat as new requests are added.

In test_triggerer_job.py multiple tests failed on client as it was not seeing the client. What ended up being the problem is that triggerer_job_runner.py was lazy initializing the client and was only creating a client when it strictly needed it, but because attrs was causing some conflicts with it, the client was being set to None causing failures. The solution ended up being removing the lazy initialization, creating the client inside the start and then updating the tests cases to align with the new specification for the client being set in WatchedSubProcess instead of the individual processesors.

Key Changes

• Registry Design Pattern: implemented a centralized registry inside request_handlers.py. Shared handlers are now used through the decorator.
• Default Dispatch: Replaced the previous raise NotImplementedError() with the registry system for WatchedSubProcess.
• Subclass Architecture: Subclasses now simply call super() for shared messages, and for any supervisor specific special cases they are still handled in-line.
• Test Suites: Updated test_integration.py, and test_triggerer_job.py test cases updated to align with the new specification for client being set in WatchedSubProcess instead in individual processors.

Handlers Moved

Moved the following handlers to the registry:
GetAssetByName
GetAssetByUri
GetAssetsByAlias
GetAssetEventByAsset
GetAssetEventByAssetAlias
TriggerDagRun
GetDagRun
GetDag
GetTaskRescheduleStartDate
GetTaskBreadcrumbs
ValidateInletsAndOutlets
CreateHITLDetailPayload
GetTaskStore
SetTaskStore
DeleteTaskStore
ClearTaskStore
Asset store CRUD:
GetAssetStoreByName
GetAssetStoreByUri
SetAssetStoreByName
SetAssetStoreByUri
DeleteAssetStoreByName
DeleteAssetStoreByUri
ClearAssetStoreByName
ClearAssetStoreByUri

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

• Yes (please specify the tool below)

Generated-by: [Sonnet 4.6] following the guidelines

---

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

---

Important

🛠️ Maintainer triage note for @JH-A-Kim · by @potiuk · 2026-07-02 17:46 UTC

Some review feedback from @ferruzzi is waiting on you:

• 1 unresolved review thread(s) from @ferruzzi need a reply or a fix.

The ball is in your court — you've been assigned to this PR. Reply or push a fix in each thread, then mark them resolved.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[potiuk]

@JH-A-Kim — converting this to draft because it currently has merge conflicts with main that need resolving before a maintainer can review. Once you've rebased onto the latest main and pushed, please hit Ready for review again and it'll re-enter the queue. Thanks for the contribution!

See the PR quality criteria.

_{This is an automated first-pass triage note drafted by an AI-assisted tool and may get things wrong; once you've addressed it, a real Apache Airflow maintainer takes the next look. (why the first pass is automated)}

---

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

[JH-A-Kim]

@ashb Would it be possible to get a review? Thank you!

[JH-A-Kim]

@ashb Would i be able to get a review? Thanks!