kaxil commented on code in PR #65543:
URL: https://github.com/apache/airflow/pull/65543#discussion_r3118690521
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -535,13 +547,21 @@ def _get_priority_files(self, session: Session =
NEW_SESSION) -> list[DagFileInf
session.delete(request)
return files
+ def fetch_callbacks(self) -> list[CallbackRequest]:
Review Comment:
`fetch_callbacks` isn't covered directly. A revert to
`self._fetch_callbacks()` on line 462 would still pass the existing tests since
the default just delegates. Worth a one-liner test that asserts
`fetch_callbacks()` returns what `_fetch_callbacks()` returns, so the seam
can't regress silently.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -569,24 +589,46 @@ def _fetch_callbacks(
guard.commit()
return callback_queue
- def _add_callback_to_queue(self, request: CallbackRequest):
- self.log.debug("Queuing %s CallbackRequest: %s",
type(request).__name__, request)
+ def resolve_callback_bundle(self, request: CallbackRequest) ->
BaseDagBundle:
+ """
+ Resolve the bundle required to execute a callback request.
+
+ Raises :class:`CallbackBundleUnavailable` if the bundle cannot be
resolved.
+ Override to source the bundle from an API.
+ """
try:
- bundle = DagBundlesManager().get_bundle(name=request.bundle_name,
version=request.bundle_version)
- except ValueError:
- # Bundle no longer configured
+ return DagBundlesManager().get_bundle(name=request.bundle_name,
version=request.bundle_version)
+ except ValueError as e:
self.log.error("Bundle %s no longer configured, skipping
callback", request.bundle_name)
- return None
+ raise CallbackBundleUnavailable(f"Bundle {request.bundle_name!r}
no longer configured") from e
+
+ def initialize_callback_bundle(self, request: CallbackRequest, bundle:
BaseDagBundle) -> None:
+ """
+ Prepare ``bundle`` so the callback can run against it.
+
+ Raises :class:`CallbackBundleUnavailable` if preparation fails and the
callback should
+ be skipped. Override to change how bundles are materialized (for
example, via an API).
+ """
if bundle.supports_versioning and request.bundle_version:
try:
bundle.initialize()
- except Exception:
+ except Exception as e:
Review Comment:
If a subclass override calls back into
`super().initialize_callback_bundle(...)` or chains to another helper that
raises `CallbackBundleUnavailable`, this `except Exception` catches it and
re-wraps with the generic "Failed to initialize bundle..." message, losing the
original reason. Consider narrowing to the expected exception types or letting
`CallbackBundleUnavailable` pass through.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -569,24 +589,46 @@ def _fetch_callbacks(
guard.commit()
return callback_queue
- def _add_callback_to_queue(self, request: CallbackRequest):
- self.log.debug("Queuing %s CallbackRequest: %s",
type(request).__name__, request)
+ def resolve_callback_bundle(self, request: CallbackRequest) ->
BaseDagBundle:
+ """
+ Resolve the bundle required to execute a callback request.
+
+ Raises :class:`CallbackBundleUnavailable` if the bundle cannot be
resolved.
+ Override to source the bundle from an API.
+ """
try:
- bundle = DagBundlesManager().get_bundle(name=request.bundle_name,
version=request.bundle_version)
- except ValueError:
- # Bundle no longer configured
+ return DagBundlesManager().get_bundle(name=request.bundle_name,
version=request.bundle_version)
+ except ValueError as e:
self.log.error("Bundle %s no longer configured, skipping
callback", request.bundle_name)
- return None
+ raise CallbackBundleUnavailable(f"Bundle {request.bundle_name!r}
no longer configured") from e
+
+ def initialize_callback_bundle(self, request: CallbackRequest, bundle:
BaseDagBundle) -> None:
+ """
+ Prepare ``bundle`` so the callback can run against it.
+
+ Raises :class:`CallbackBundleUnavailable` if preparation fails and the
callback should
+ be skipped. Override to change how bundles are materialized (for
example, via an API).
+ """
Review Comment:
The docstring says "prepare bundle so the callback can run" but when
`supports_versioning` is False or `bundle_version` is None, this silently
no-ops. That's deliberate for the DB-backed flow, but an override author
reading only the docstring may not realize they also inherit the
skip-when-unversioned behavior. Could be explicit: "Default implementation is a
no-op for unversioned bundles or requests without a version."
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -569,24 +589,46 @@ def _fetch_callbacks(
guard.commit()
return callback_queue
- def _add_callback_to_queue(self, request: CallbackRequest):
- self.log.debug("Queuing %s CallbackRequest: %s",
type(request).__name__, request)
+ def resolve_callback_bundle(self, request: CallbackRequest) ->
BaseDagBundle:
+ """
+ Resolve the bundle required to execute a callback request.
+
+ Raises :class:`CallbackBundleUnavailable` if the bundle cannot be
resolved.
+ Override to source the bundle from an API.
+ """
try:
- bundle = DagBundlesManager().get_bundle(name=request.bundle_name,
version=request.bundle_version)
- except ValueError:
- # Bundle no longer configured
+ return DagBundlesManager().get_bundle(name=request.bundle_name,
version=request.bundle_version)
+ except ValueError as e:
self.log.error("Bundle %s no longer configured, skipping
callback", request.bundle_name)
- return None
+ raise CallbackBundleUnavailable(f"Bundle {request.bundle_name!r}
no longer configured") from e
+
+ def initialize_callback_bundle(self, request: CallbackRequest, bundle:
BaseDagBundle) -> None:
+ """
+ Prepare ``bundle`` so the callback can run against it.
+
+ Raises :class:`CallbackBundleUnavailable` if preparation fails and the
callback should
+ be skipped. Override to change how bundles are materialized (for
example, via an API).
+ """
if bundle.supports_versioning and request.bundle_version:
try:
bundle.initialize()
- except Exception:
+ except Exception as e:
self.log.exception(
"Error initializing bundle %s version %s for callback,
skipping",
request.bundle_name,
request.bundle_version,
)
- return None
+ raise CallbackBundleUnavailable(
+ f"Failed to initialize bundle {request.bundle_name!r}
version {request.bundle_version!r}"
+ ) from e
+
+ def _add_callback_to_queue(self, request: CallbackRequest) -> None:
+ self.log.debug("Queuing %s CallbackRequest: %s",
type(request).__name__, request)
+ try:
+ bundle = self.resolve_callback_bundle(request)
+ self.initialize_callback_bundle(request, bundle)
+ except CallbackBundleUnavailable:
Review Comment:
The swallow here relies on raise sites logging before raising, which both
default impls do. If an API-backed override raises `CallbackBundleUnavailable`
without logging, the callback will be silently dropped. A
`self.log.debug("Skipping callback %s: %s", request, exc)` here would make the
contract self-enforcing.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1565,10 +1641,30 @@ def
test_add_callback_initializes_versioned_bundle(self, mock_bundle_manager):
bundle.initialize.assert_called_once()
Review Comment:
This asserts `bundle.initialize` was called but not that
`_callback_to_execute` was actually populated. Pre-existing gap, but now
asymmetric with the new skip-case tests below which do assert the negative.
Adding `assert manager._callback_to_execute` here would balance the coverage.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
