You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ephraimbuddy (via GitHub)" <gi...@apache.org> on 2026/04/29 15:49:44 UTC
[PR] Add purge-warnings and bundle-refresh override seams to DagFileProcessorManager [airflow]
ephraimbuddy opened a new pull request, #66107:
URL: https://github.com/apache/airflow/pull/66107
The projected AIP-92 DB-to-API swap needs DagFileProcessorManager subclasses that route operations through the API server instead of the metadata DB. Two spots in the current implementation are awkward to override cleanly:
- _run_parsing_loop calls DagWarning.purge_inactive_dag_warnings() directly, forcing subclasses to reimplement the parsing loop just to redirect or skip the cleanup. Promote it to an overridable instance method, purge_inactive_dag_warnings(), so subclasses can forward it to an API or no-op it without touching _run_parsing_loop.
- Bundle refreshes triggered by external events (API callbacks, coordinator messages) had to mutate the private _force_refresh_bundles set directly. Expose request_bundle_refresh() as the public seam so event handlers can mark a bundle for refresh without reaching into private state.
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes (please specify the tool below)
Generated-by: Claude code (opus 4.7)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add purge-warnings and bundle-refresh override seams to DagFileProcessorManager [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #66107:
URL: https://github.com/apache/airflow/pull/66107#discussion_r3166860723
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -534,6 +534,16 @@ def claim_priority_files(self) -> list[DagFileInfo]:
"""
return self._claim_priority_files()
+ def request_bundle_refresh(self, bundle_name: str) -> None:
+ """
+ Request that ``bundle_name`` be refreshed on the next refresh tick.
+
+ Use this from event handlers reacting to external signals to mark a
+ bundle as needing refresh; the next call to :meth:`_refresh_dag_bundles`
+ will not skip it via :meth:`should_skip_refresh`.
+ """
+ self._force_refresh_bundles.add(bundle_name)
Review Comment:
Looks like this not currently used anywhere?
Since the only existing usage to `_force_refresh_bundles` is in `_queue_requested_files_for_parsing`, and it actually bulk-adds multiple names at one time, should this function handle multiple names instead?
```python
def request_bundle_refresh(self, bundle_names: Iterable[str]) -> None:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add purge-warnings and bundle-refresh override seams to DagFileProcessorManager [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #66107:
URL: https://github.com/apache/airflow/pull/66107#discussion_r3182263298
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -691,6 +701,15 @@ def update_bundle_state(
values["version"] = version
session.execute(update(DagBundleModel).where(DagBundleModel.name == bundle_name).values(**values))
+ def purge_inactive_dag_warnings(self) -> None:
+ """
+ Purge expired DAG warnings.
+
+ Default implementation deletes records from the metadata DB; override to
Review Comment:
Docstring says "Purge expired DAG warnings", but the delegated implementation (`DagWarning.purge_inactive_dag_warnings`) purges warnings for *inactive/stale* DAGs (not time-expired warnings). Please adjust the wording to match the actual behavior to avoid confusion for subclass authors overriding this seam.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -537,6 +537,16 @@ def claim_priority_files(self) -> list[DagFileInfo]:
"""
return self._claim_priority_files()
+ def request_bundle_refresh(self, bundle_names: Iterable[str]) -> None:
+ """
+ Request that the given bundles be refreshed on the next refresh tick.
+
+ Use this from event handlers reacting to external signals to mark
+ bundles as needing refresh; the next call to :meth:`_refresh_dag_bundles`
+ will not skip them via :meth:`should_skip_refresh`.
+ """
+ self._force_refresh_bundles.update(bundle_names)
Review Comment:
`request_bundle_refresh()` accepts an `Iterable[str]`, so if a caller accidentally passes a single bundle name as a plain string (e.g. `"bundleone"`) it will be treated as an iterable of characters and add `{'b','u',...}` to `_force_refresh_bundles`. Since this is a new public seam intended for external event handlers, it would be safer to either (a) accept `bundle_name: str | Iterable[str]` and special-case `str`, or (b) validate/raise when `bundle_names` is a `str` to prevent hard-to-debug behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add purge-warnings and bundle-refresh override seams to DagFileProcessorManager [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy merged PR #66107:
URL: https://github.com/apache/airflow/pull/66107
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org