ephraimbuddy commented on code in PR #65543:
URL: https://github.com/apache/airflow/pull/65543#discussion_r3119153745
##########
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:
Removed the exception
--
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]
