You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "o-nikolas (via GitHub)" <gi...@apache.org> on 2026/06/30 16:35:02 UTC
[PR] Thread version_data to callbacks [airflow]
o-nikolas opened a new pull request, #69185:
URL: https://github.com/apache/airflow/pull/69185
PR #67217 wired version_data into the task execution path but not callbacks, so callbacks for a pinned run initialized their bundle without the manifest tasks used. Populate version_data on the callback producer paths (ExecuteCallback.make and the CallbackRequest creation sites) under the same pin guard as tasks, carry it on BaseCallbackRequest, and forward it through prepare_callback_bundle to get_bundle.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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] Thread version_data to callbacks [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #69185:
URL: https://github.com/apache/airflow/pull/69185#discussion_r3507374269
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -235,3 +235,21 @@ def get_version(
def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+
+
+def resolve_pinned_version_data(
+ dag_version: DagVersion | None, bundle_version: str | None
+) -> dict[str, Any] | None:
+ """
+ Return a bundle version's ``version_data`` manifest, but only for pinned runs.
Review Comment:
Sam here. I would keep the first line. Leave any detailed explanations for the comments.
##########
airflow-core/src/airflow/callbacks/callback_requests.py:
##########
@@ -46,6 +46,12 @@ class BaseCallbackRequest(BaseModel):
"""File Path to use to run the callback"""
bundle_name: str
bundle_version: str | None
+ version_data: dict[str, Any] | None = None
+ """Optional structured metadata for the pinned bundle version (e.g. an S3 object manifest).
+
+ Populated only for pinned runs so the callback initializes the bundle against the same
+ version the task ran with; ``None`` for unpinned runs.
Review Comment:
I would keep only the first line. The docstring under each parameter is only intended to explain what the field is. Not its inner workings.
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -235,3 +235,21 @@ def get_version(
def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+
+
+def resolve_pinned_version_data(
Review Comment:
Minor nit: should this be `_resolve_pinned_version_data`? It looks like an internal helper that's only used within Airflow itself, so I'm not sure it needs to be part of the module's public API. I would prefer `_resolve_version_data` to be more concise.
--
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] Thread version_data to callbacks [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #69185:
URL: https://github.com/apache/airflow/pull/69185#discussion_r3532250165
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -1967,7 +1968,28 @@ def test_prepare_callback_bundle_initializes_versioned_bundle(self, mock_bundle_
bundle.initialize.assert_called_once()
@mock.patch("airflow.dag_processing.manager.DagBundlesManager")
- def test_prepare_callback_bundle_skips_initialize_for_unversioned_request(self, mock_bundle_manager):
+ def test_prepare_callback_bundle_forwards_version_data(self, mock_bundle_manager):
+ manager = DagFileProcessorManager(max_runs=1)
+ bundle = MagicMock(spec=BaseDagBundle)
+ bundle.supports_versioning = True
+ mock_bundle_manager.return_value.get_bundle.return_value = bundle
+
+ version_data = {"schema_version": 1, "files": {"dags/my_dag.py": "ver123"}}
+ request = DagCallbackRequest(
+ filepath="file1.py",
+ dag_id="dag1",
+ run_id="run1",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version="some_commit_hash",
+ version_data=version_data,
+ msg=None,
+ )
+
+ manager.prepare_callback_bundle(request)
+ mock_bundle_manager.return_value.get_bundle.assert_called_once_with(
+ name="testing", version="some_commit_hash", version_data=version_data
+ )
manager = DagFileProcessorManager(max_runs=1)
Review Comment:
Good catch, will fix
--
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] Thread version_data to callbacks [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #69185:
URL: https://github.com/apache/airflow/pull/69185#discussion_r3532238012
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -235,3 +235,21 @@ def get_version(
def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+
+
+def resolve_pinned_version_data(
Review Comment:
Perhaps, underscore has really muddied meaning in Python, and is barely respected anyway. But I also don't mind either way, happy to change it if you prefer.
--
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] Thread version_data to callbacks [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #69185:
URL: https://github.com/apache/airflow/pull/69185#discussion_r3532249079
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -570,12 +566,18 @@ def _submit_callback_if_necessary() -> None:
if event.task_instance_state in (TaskInstanceState.SUCCESS, TaskInstanceState.FAILED):
if task_instance.dag_model.relative_fileloc is None:
raise RuntimeError("relative_fileloc should not be None for a finished task")
+ from airflow.models.dag_version import resolve_pinned_version_data
+
+ version_data = resolve_pinned_version_data(
+ task_instance.dag_version, task_instance.dag_run.bundle_version
Review Comment:
Yeah, good catch, I'll try to sync them
--
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] Thread version_data to callbacks [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #69185:
URL: https://github.com/apache/airflow/pull/69185#discussion_r3532311005
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -235,3 +235,21 @@ def get_version(
def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+
+
+def resolve_pinned_version_data(
Review Comment:
Are these functions intended to be public APIs or internal helpers? I think it is important to be able to easily distinguish between both. I think underscoring is just an convenient way to enable that without having to infer whether a function is one or the other by tracing its usage across the codebase.
--
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] Thread version_data to callbacks [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #69185:
URL: https://github.com/apache/airflow/pull/69185
--
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] Thread version_data to callbacks [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #69185:
URL: https://github.com/apache/airflow/pull/69185#discussion_r3532275473
##########
airflow-core/src/airflow/callbacks/callback_requests.py:
##########
@@ -46,6 +46,12 @@ class BaseCallbackRequest(BaseModel):
"""File Path to use to run the callback"""
bundle_name: str
bundle_version: str | None
+ version_data: dict[str, Any] | None = None
+ """Optional structured metadata for the pinned bundle version (e.g. an S3 object manifest).
+
+ Populated only for pinned runs so the callback initializes the bundle against the same
+ version the task ran with; ``None`` for unpinned runs.
Review Comment:
I agree with you. I generally err in favour of over-documenting but I believe docstrings should not be too verbose because the extra context just gets cutoff in IDE tooltips. I think you should keep the the 2 lines after the first as a comment instead.
But this is not a major issue. It is up to you.
--
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] Thread version_data to callbacks [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #69185:
URL: https://github.com/apache/airflow/pull/69185#discussion_r3532243252
##########
airflow-core/src/airflow/callbacks/callback_requests.py:
##########
@@ -46,6 +46,12 @@ class BaseCallbackRequest(BaseModel):
"""File Path to use to run the callback"""
bundle_name: str
bundle_version: str | None
+ version_data: dict[str, Any] | None = None
+ """Optional structured metadata for the pinned bundle version (e.g. an S3 object manifest).
+
+ Populated only for pinned runs so the callback initializes the bundle against the same
+ version the task ran with; ``None`` for unpinned runs.
Review Comment:
I suppose yeah, I'll change it. I am worried about the recent push to strip helpful context/information from our codebase though. These bits of context are often super helpful.
--
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