Refactor bundle refresh persistence into overridable get/update methods

[ephraimbuddy]

Replace inline DagBundleModel mutations in _refresh_dag_bundles() with get_bundle_state() and update_bundle_state() instance methods on DagFileProcessorManager, isolating all DB access for bundle state behind a clean override seam.

---

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

• Yes (please specify the tool below)

Generated-by: Sonnet 4.6

[Copilot]

Pull request overview

This PR refactors DAG bundle refresh persistence in DagFileProcessorManager._refresh_dag_bundles() by moving direct DagBundleModel mutations into two new instance methods (get_bundle_state() and update_bundle_state()), creating a clean override seam for DB access and adding unit tests for the new behavior.

Changes:

• Introduce BundleState and DagFileProcessorManager.get_bundle_state() / update_bundle_state() to encapsulate persisted bundle refresh state.
• Update _refresh_dag_bundles() to use the new methods (including error-handling paths) instead of inline session usage.
• Add unit tests covering state read/write behavior and refresh loop interactions via mocked get/update methods.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
airflow-core/src/airflow/dag_processing/manager.py	Adds BundleState + get/update persistence methods; refactors _refresh_dag_bundles() to call them.
airflow-core/tests/unit/dag_processing/test_manager.py	Adds unit tests for the new persistence methods and refresh loop behavior.

[Copilot]

Pull request overview

This PR refactors how DagFileProcessorManager._refresh_dag_bundles() reads/writes persisted DAG bundle refresh state by introducing overridable get_bundle_state() / update_bundle_state() methods, providing a cleaner seam for isolating DB access.

Changes:

• Add a BundleState type plus get_bundle_state() / update_bundle_state() methods on DagFileProcessorManager.
• Refactor _refresh_dag_bundles() to use those methods (and to log/continue on state read/write failures).
• Add unit tests covering the new methods and key _refresh_dag_bundles() state/persistence behaviors.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
airflow-core/src/airflow/dag_processing/manager.py	Introduces BundleState and new DB access methods; updates bundle refresh loop to use them.
airflow-core/tests/unit/dag_processing/test_manager.py	Adds unit tests for bundle state get/update and refresh persistence paths.